Oswaldo L. V. Costa

"""APK Code Analysis Extractor.

Decompiles and disassembles APK DEX bytecode at the method level,
producing per-method content and reference records analogous to
the Binary Ninja code_binja_* tables.

Uses androguard + JADX + apktool to replicate Binary Ninja analysis
depth for Android applications.
"""

import gc
import hashlib
import inspect
import logging
import os
import threading
import time
from datetime import datetime, timezone
from typing import Any, Dict, Optional

from redb.extractors.decompiler.apk.analyzer import APKCodeAnalyzer
from redb.extractors.enum import Tag
from redb.extractors.extractor import Extractor


class DecompileAPK(Extractor):
    """APK code analysis extractor — produces multi-table ClickHouse export.

    Follows the same pattern as DecompileBinja for consistency.
    """

    def __init__(
        self,
        filepath,
        log,
        exporters=None,
        index_prefix=None,
        known_benign=False,
        known_malicious=False,
        filetype=None,
        decompile_modules=None,
    ):
        super().__init__(
            filepath,
            log,
            exporters,
            index_prefix,
            known_benign=known_benign,
            known_malicious=known_malicious,
        )
        self.log.debug(inspect.currentframe().f_code.co_name)
        self.analysis_results = None
        self.analyzer = None
        self.filetype = filetype or "apk"
        self.decompile_modules = decompile_modules or {"all"}

        try:
            self.APK_DECOMPILE_TIMEOUT = int(
                os.getenv("APK_DECOMPILE_TIMEOUT", "600")
            )
        except ValueError:
            self.log.warning(
                "Invalid APK_DECOMPILE_TIMEOUT value, using default of 600 seconds"
            )
            self.APK_DECOMPILE_TIMEOUT = 600

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.cleanup_run()

    def calculate_md5(self, input_str):
        """Calculate MD5 hash of a string."""
        return hashlib.md5(input_str.encode("utf-8")).hexdigest()

    def cleanup_run(self):
        """Clean up after analysis."""
        try:
            if self.analyzer:
                self.analyzer.cleanup()
                self.analyzer = None
            gc.collect()
        except Exception as e:
            self.log.error(f"Error in cleanup: {e}")

    def analyze_apk(self) -> Optional[Dict[str, Any]]:
        """Run APK code analysis and return results."""
        self.log.debug("Starting APK code analysis")
        try:
            self.analyzer = APKCodeAnalyzer(
                filepath=self.filepath,
                timeout=self.APK_DECOMPILE_TIMEOUT,
                log=self.log,
                decompile_modules=self.decompile_modules,
            )
            results = self.analyzer.extract()
            return results
        except Exception as e:
            self.log.error(f"Error in APK code analysis: {e}")
            import traceback
            self.log.error(f"Traceback: {traceback.format_exc()}")
            return None
        finally:
            self.cleanup_run()

    def extract(self):
        """Extract and process all analysis results.

        Uses daemon thread with timeout, same pattern as DecompileBinja.
        """
        self.log.debug(inspect.currentframe().f_code.co_name)

        extraction_completed = False
        extraction_result = False
        extraction_error = None

        def do_extraction():
            nonlocal extraction_completed, extraction_result, extraction_error
            try:
                results = self.analyze_apk()
                if not results:
                    extraction_result = False
                else:
                    self.analysis_results = results
                    self.analysis_results["sha256"] = self.sha256
                    self.analysis_results["sha1"] = self.sha1
                    self.analysis_results["md5"] = self.md5
                    extraction_result = True
            except Exception as e:
                extraction_error = e
                extraction_result = False
            finally:
                extraction_completed = True

        extraction_thread = threading.Thread(target=do_extraction)
        extraction_thread.daemon = True
        extraction_thread.start()

        start_time = time.time()
        while (
            not extraction_completed
            and (time.time() - start_time) < self.APK_DECOMPILE_TIMEOUT
        ):
            time.sleep(1)

        if not extraction_completed:
            self.log.error(
                f"APK extraction timed out after {self.APK_DECOMPILE_TIMEOUT} seconds"
            )
            self.cleanup_run()
            return None

        if extraction_error:
            self.log.error(f"Error in APK extraction: {extraction_error}")
            return None

        return self.analysis_results if extraction_result else None

    def prepare_export_data(self, exporter_type: str) -> Any:
        """Prepare data for database export."""
        self.log.debug(inspect.currentframe().f_code.co_name)
        if not self.analysis_results:
            return None

        if exporter_type == "ClickHouseExporter":
            now = datetime.now(timezone.utc)
            export = {"multi_table": True}

            # Table 1: Decompiled method content
            if self.analysis_results.get("decompiled_content"):
                export["decompiled_content"] = {
                    "table": "code_apk_decompiled_methods_content",
                    "data": [
                        [
                            f["decompiled_method_hash"],
                            f["decompiled_method"],
                            f.get("decompiled_method_type", "UNKNOWN"),
                            1 if f.get("decompiled_has_string_encryption") else 0,
                            1 if f.get("decompiled_has_reflection_calls") else 0,
                            1 if f.get("decompiled_excessive_goto_count") else 0,
                            now,
                        ]
                        for f in self.analysis_results["decompiled_content"]
                    ],
                    "column_names": [
                        "decompiled_method_hash",
                        "decompiled_method",
                        "decompiled_method_type",
                        "decompiled_has_string_encryption",
                        "decompiled_has_reflection_calls",
                        "decompiled_excessive_goto_count",
                        "analysis_date",
                    ],
                    "column_type_names": [
                        "FixedString(64)",
                        "String",
                        "Enum8('USER'=1, 'LIBRARY'=2, 'UNKNOWN'=5)",
                        "UInt8",
                        "UInt8",
                        "UInt8",
                        "DateTime64(3, 'UTC')",
                    ],
                }

            # Table 2: Decompiled method references
            if self.analysis_results.get("decompiled_refs"):
                export["decompiled_refs"] = {
                    "table": "code_apk_decompiled_methods_references",
                    "data": [
                        [
                            self.analysis_results["sha256"],
                            f["decompiled_method_hash"],
                            f.get("smali_method_hash"),
                            f.get("decompiled_class_name", ""),
                            f.get("decompiled_method_name", ""),
                            f.get("decompiled_method_signature", ""),
                            f.get("decompiled_method_prototype", ""),
                            f.get("functions_caller", []),
                            f.get("functions_call", []),
                            now,
                        ]
                        for f in self.analysis_results["decompiled_refs"]
                    ],
                    "column_names": [
                        "sha256",
                        "decompiled_method_hash",
                        "smali_method_hash",
                        "decompiled_class_name",
                        "decompiled_method_name",
                        "decompiled_method_signature",
                        "decompiled_method_prototype",
                        "functions_caller",
                        "functions_call",
                        "analysis_date",
                    ],
                    "column_type_names": [
                        "FixedString(64)",
                        "FixedString(64)",
                        "Nullable(FixedString(64))",
                        "LowCardinality(String)",
                        "LowCardinality(String)",
                        "String",
                        "String",
                        "Array(String)",
                        "Array(String)",
                        "DateTime64(3, 'UTC')",
                    ],
                }

            # Table 3: Smali method content
            if self.analysis_results.get("smali_content"):
                export["smali_content"] = {
                    "table": "code_apk_smali_methods_content",
                    "data": [
                        [
                            f["smali_method_hash"],
                            f["smali_method"],
                            f.get("smali_method_type", "UNKNOWN"),
                            f.get("smali_instructions_count", 0),
                            f.get("smali_register_count", 0),
                            1 if f.get("smali_has_string_encryption") else 0,
                            1 if f.get("smali_has_reflection_calls") else 0,
                            1 if f.get("smali_excessive_goto_count") else 0,
                            f.get("smali_flattened_score", 0.0),
                            f.get("smali_mba_score", 0.0),
                            now,
                        ]
                        for f in self.analysis_results["smali_content"]
                    ],
                    "column_names": [
                        "smali_method_hash",
                        "smali_method",
                        "smali_method_type",
                        "smali_instructions_count",
                        "smali_register_count",
                        "smali_has_string_encryption",
                        "smali_has_reflection_calls",
                        "smali_excessive_goto_count",
                        "smali_flattened_score",
                        "smali_mba_score",
                        "analysis_date",
                    ],
                    "column_type_names": [
                        "FixedString(64)",
                        "String",
                        "Enum8('USER'=1, 'LIBRARY'=2, 'UNKNOWN'=5)",
                        "UInt32",
                        "UInt16",
                        "UInt8",
                        "UInt8",
                        "UInt8",
                        "Float64",
                        "Float64",
                        "DateTime64(3, 'UTC')",
                    ],
                }

            # Table 4: Smali method references
            if self.analysis_results.get("smali_refs"):
                export["smali_refs"] = {
                    "table": "code_apk_smali_methods_references",
                    "data": [
                        [
                            self.analysis_results["sha256"],
                            f["smali_method_hash"],
                            f.get("decompiled_method_hash"),
                            f.get("smali_class_name", ""),
                            f.get("smali_method_name", ""),
                            f.get("smali_method_signature", ""),
                            now,
                        ]
                        for f in self.analysis_results["smali_refs"]
                    ],
                    "column_names": [
                        "sha256",
                        "smali_method_hash",
                        "decompiled_method_hash",
                        "smali_class_name",
                        "smali_method_name",
                        "smali_method_signature",
                        "analysis_date",
                    ],
                    "column_type_names": [
                        "FixedString(64)",
                        "FixedString(64)",
                        "Nullable(FixedString(64))",
                        "LowCardinality(String)",
                        "LowCardinality(String)",
                        "String",
                        "DateTime64(3, 'UTC')",
                    ],
                }

            # Table 5: Method similarity metrics (content-based fuzzy matching)
            if self.analysis_results.get("similarity_metrics"):
                export["method_similarity_metrics"] = {
                    "table": "code_apk_method_similarity_metrics",
                    "data": [
                        [
                            f["smali_method_hash"],
                            f.get("ssdeep_smali"),
                            f.get("tlsh_smali"),
                            f.get("ssdeep_smali_normalized"),
                            f.get("tlsh_smali_normalized"),
                            f.get("minhash", []),
                            now,
                        ]
                        for f in self.analysis_results["similarity_metrics"]
                    ],
                    "column_names": [
                        "smali_method_hash",
                        "ssdeep_smali",
                        "tlsh_smali",
                        "ssdeep_smali_normalized",
                        "tlsh_smali_normalized",
                        "minhash",
                        "analysis_date",
                    ],
                    "column_type_names": [
                        "FixedString(64)",
                        "Nullable(String)",
                        "Nullable(FixedString(72))",
                        "Nullable(String)",
                        "Nullable(FixedString(72))",
                        "Array(UInt8)",
                        "DateTime64(3, 'UTC')",
                    ],
                }

            # Table 5b: CFG method features (structural/topological)
            if self.analysis_results.get("cfg"):
                export["cfg_methods"] = {
                    "table": "code_apk_cfg_methods",
                    "data": [
                        [
                            cfg["smali_method_hash"],
                            cfg["cfg_topology_hash"],
                            cfg["block_count"],
                            cfg["edge_count"],
                            cfg.get("cfg_instructions_count", 0),
                            cfg.get("call_count", 0),
                            cfg["cyclomatic_complexity"],
                            cfg.get("loop_count", 0),
                            cfg.get("max_depth", 0),
                            cfg.get("max_fan_out", 0),
                            cfg.get("md_index_topdown", 0),
                            cfg.get("md_index_bottomup", 0),
                            cfg.get("prime_product_smali", 0),
                            cfg.get("cfg_feature_tlsh"),
                            cfg.get("wl_minhash", []),
                            cfg.get("bb_features", []),
                            cfg.get("cfg_adjacency", []),
                            now,
                        ]
                        for cfg in self.analysis_results["cfg"]
                        if cfg is not None
                    ],
                    "column_names": [
                        "smali_method_hash",
                        "cfg_topology_hash",
                        "block_count",
                        "edge_count",
                        "cfg_instructions_count",
                        "call_count",
                        "cyclomatic_complexity",
                        "loop_count",
                        "max_depth",
                        "max_fan_out",
                        "md_index_topdown",
                        "md_index_bottomup",
                        "prime_product_smali",
                        "cfg_feature_tlsh",
                        "wl_minhash",
                        "bb_features",
                        "cfg_adjacency",
                        "analysis_date",
                    ],
                    "column_type_names": [
                        "FixedString(64)",
                        "FixedString(16)",
                        "UInt16",
                        "UInt16",
                        "UInt32",
                        "UInt16",
                        "UInt16",
                        "UInt16",
                        "UInt16",
                        "UInt16",
                        "UInt64",
                        "UInt64",
                        "UInt64",
                        "Nullable(FixedString(72))",
                        "Array(UInt8)",
                        "Array(Array(UInt16))",
                        "Array(UInt32)",
                        "DateTime64(3, 'UTC')",
                    ],
                }

            # Table 6: Strings — reuse code_binja_strings_raw for cross-format correlation
            # DEX strings are MUTF-8; string_raw = string since no encoding difference
            if self.analysis_results.get("strings"):
                export["strings_raw"] = {
                    "table": "code_binja_strings_raw",
                    "data": [
                        [
                            self.analysis_results["sha256"],
                            s["string"],
                            s["string"],  # string_raw = string (MUTF-8 decoded to UTF-8)
                            s.get("string_encoding", "UTF8"),
                            s.get("string_offset", 0),
                            s.get("string_length", len(s["string"])),
                            s.get("string_length", len(s["string"])),  # string_raw_length = string_length
                            s.get("string_entropy", 0.0),
                        ]
                        for s in self.analysis_results["strings"]
                    ],
                    "column_names": [
                        "sha256",
                        "string",
                        "string_raw",
                        "string_encoding",
                        "string_offset",
                        "string_length",
                        "string_raw_length",
                        "string_entropy",
                    ],
                    "column_type_names": [
                        "FixedString(64)",
                        "String",
                        "String",
                        "LowCardinality(String)",
                        "UInt64",
                        "UInt32",
                        "UInt32",
                        "Float32",
                    ],
                }

            # Table 7: Analysis errors
            if self.analysis_results.get("analysis_errors"):
                export["analysis_errors"] = {
                    "table": "code_apk_analysis_errors",
                    "data": [
                        [
                            self.analysis_results["sha256"],
                            f.get("class_name"),
                            f.get("method_name"),
                            f.get("error_location", "unknown"),
                            f.get("error_message", ""),
                            f.get("error_type", "unknown"),
                            self.calculate_md5(
                                f"{f.get('error_message', '')}"
                                f"{f.get('class_name', '')}"
                                f"{f.get('method_name', '')}"
                                f"{f.get('error_location', 'unknown')}"
                            ),
                            "new",
                            now,
                        ]
                        for f in self.analysis_results["analysis_errors"]
                    ],
                    "column_names": [
                        "sha256",
                        "class_name",
                        "method_name",
                        "error_location",
                        "error_message",
                        "error_type",
                        "error_hash",
                        "status",
                        "analysis_date",
                    ],
                    "column_type_names": [
                        "FixedString(64)",
                        "Nullable(String)",
                        "Nullable(String)",
                        "LowCardinality(String)",
                        "Nullable(String)",
                        "Nullable(String)",
                        "FixedString(32)",
                        "Enum8('new'=1, 'investigating'=2, 'fixed'=3, 'wontfix'=4)",
                        "DateTime64(3, 'UTC')",
                    ],
                }

            return export

        return None

    def tag(self) -> str:
        """Return the tag for this extractor."""
        return Tag.APK_DECOMPILED.value

    def get_clickhouse_table(self) -> str:
        """Not used directly as we're handling multiple tables."""
        pass