Vincent J. van Heuven

import hashlib
import inspect
import json
from collections import Counter
from datetime import datetime, timezone
from typing import Any

import tlsh

from redb.extractors.enum import Tag
from redb.extractors.apk_extractor import APKExtractor
from redb.models.dataclasses import APKDexFile

# Sensitive API patterns categorized for malware analysis
SENSITIVE_API_CATEGORIES = {
    "reflection": [
        "Ljava/lang/reflect/Method;->invoke",
        "Ljava/lang/reflect/Field;->get",
        "Ljava/lang/reflect/Field;->set",
        "Ljava/lang/reflect/Constructor;->newInstance",
        "Ljava/lang/Class;->forName",
        "Ljava/lang/Class;->getMethod",
        "Ljava/lang/Class;->getDeclaredMethod",
        "Ljava/lang/Class;->getDeclaredField",
        "Ljava/lang/ClassLoader;->loadClass",
    ],
    "crypto": [
        "Ljavax/crypto/Cipher;->getInstance",
        "Ljavax/crypto/Cipher;->init",
        "Ljavax/crypto/spec/SecretKeySpec;-><init>",
        "Ljavax/crypto/spec/IvParameterSpec;-><init>",
        "Ljava/security/MessageDigest;->getInstance",
        "Ljava/security/KeyStore;->getInstance",
        "Ljavax/crypto/Mac;->getInstance",
    ],
    "dynamic_loading": [
        "Ldalvik/system/DexClassLoader;-><init>",
        "Ldalvik/system/PathClassLoader;-><init>",
        "Ldalvik/system/InMemoryDexClassLoader;-><init>",
        "Ldalvik/system/BaseDexClassLoader;-><init>",
        "Ljava/lang/Runtime;->exec",
        "Ljava/lang/ProcessBuilder;->start",
    ],
    "telephony": [
        "Landroid/telephony/TelephonyManager;->getDeviceId",
        "Landroid/telephony/TelephonyManager;->getSubscriberId",
        "Landroid/telephony/TelephonyManager;->getLine1Number",
        "Landroid/telephony/TelephonyManager;->getSimSerialNumber",
        "Landroid/telephony/TelephonyManager;->getNetworkOperator",
        "Landroid/telephony/TelephonyManager;->getSimOperator",
    ],
    "sms": [
        "Landroid/telephony/SmsManager;->sendTextMessage",
        "Landroid/telephony/SmsManager;->sendMultipartTextMessage",
        "Landroid/telephony/SmsManager;->sendDataMessage",
    ],
    "network": [
        "Ljava/net/HttpURLConnection;->connect",
        "Ljava/net/URL;->openConnection",
        "Lokhttp3/OkHttpClient;-><init>",
        "Lokhttp3/Request$Builder;->build",
        "Lorg/apache/http/client/HttpClient;->execute",
        "Landroid/webkit/WebView;->loadUrl",
        "Landroid/webkit/WebView;->setWebViewClient",
    ],
    "native": [
        "Ljava/lang/System;->loadLibrary",
        "Ljava/lang/System;->load",
        "Ljava/lang/Runtime;->loadLibrary",
    ],
    "device_info": [
        "Landroid/os/Build;->FINGERPRINT",
        "Landroid/os/Build;->MODEL",
        "Landroid/os/Build;->MANUFACTURER",
        "Landroid/os/Build;->PRODUCT",
        "Landroid/os/Build;->BRAND",
        "Landroid/os/Build;->DEVICE",
        "Landroid/os/Build;->HARDWARE",
        "Landroid/os/Build;->SERIAL",
        "Landroid/os/Build$VERSION;->SDK_INT",
        "Landroid/provider/Settings$Secure;->getString",
    ],
    "file_io": [
        "Ljava/io/FileOutputStream;-><init>",
        "Ljava/io/FileInputStream;-><init>",
        "Landroid/content/SharedPreferences;->edit",
        "Landroid/database/sqlite/SQLiteDatabase;->execSQL",
        "Landroid/database/sqlite/SQLiteDatabase;->rawQuery",
    ],
    "ipc": [
        "Landroid/content/ContentResolver;->query",
        "Landroid/content/ContentResolver;->insert",
        "Landroid/content/ContentResolver;->delete",
        "Landroid/content/Intent;-><init>",
        "Landroid/content/Context;->sendBroadcast",
        "Landroid/content/Context;->startService",
        "Landroid/content/Context;->bindService",
    ],
}


class APKDexExtractor(APKExtractor):

    def __init__(
        self, filepath, log, exporters=None, index_prefix=None,
        known_benign=False, known_malicious=False,
        apk=None,
    ):
        super().__init__(
            filepath, log, exporters, index_prefix,
            known_benign, known_malicious, apk,
        )
        self.dex_files = []
        self.log.debug(inspect.currentframe().f_code.co_name)

    def tag(self):
        return Tag.APK_DEX.value

    def _analyze_api_usage(self, dex):
        """Categorize sensitive API calls found in DEX bytecode."""
        api_usage = {cat: set() for cat in SENSITIVE_API_CATEGORIES}

        try:
            for method in dex.get_encoded_methods():
                code = method.get_code()
                if not code:
                    continue
                try:
                    bytecode = code.get_bc()
                    if not bytecode:
                        continue
                    for instruction in bytecode.get_instructions():
                        op_name = instruction.get_name()
                        if not op_name or not op_name.startswith("invoke"):
                            continue
                        output = instruction.get_output()
                        if not output:
                            continue
                        for category, patterns in SENSITIVE_API_CATEGORIES.items():
                            for pattern in patterns:
                                if pattern in output:
                                    api_usage[category].add(output.strip())
                                    break
                except Exception:
                    continue
        except Exception as e:
            self.log.warning(f"Error analyzing API usage: {e}")

        # Convert sets to sorted lists
        return {cat: sorted(calls) for cat, calls in api_usage.items() if calls}

    def _compute_obfuscation_indicators(self, classes, methods):
        """Compute obfuscation indicators from class/method names."""
        class_names = []
        for cls in classes:
            try:
                name = cls.get_name()
                if name:
                    class_names.append(name)
            except Exception:
                continue

        method_names = []
        for m in methods:
            try:
                name = m.get_name()
                if name and name not in ("<init>", "<clinit>"):
                    method_names.append(name)
            except Exception:
                continue

        if not class_names:
            return None

        # Short class names: extract just the class part from Dalvik notation
        short_class = 0
        for n in class_names:
            simple = n.split("/")[-1].rstrip(";")
            if len(simple) <= 2:
                short_class += 1

        short_method = sum(1 for n in method_names if len(n) <= 2)

        return {
            "short_class_names_pct": round(short_class / max(len(class_names), 1), 4),
            "short_method_names_pct": round(short_method / max(len(method_names), 1), 4),
            "non_ascii_identifiers": sum(1 for n in class_names if not n.isascii()),
            "avg_class_name_length": round(
                sum(len(n) for n in class_names) / max(len(class_names), 1), 2
            ),
        }

    def _compute_top_packages(self, classes):
        """Compute top packages by class count."""
        package_counter = Counter()
        for cls in classes:
            try:
                name = cls.get_name()  # "Lcom/example/foo/Bar;"
                if not name:
                    continue
                parts = name[1:].replace("/", ".").rsplit(".", 1)
                if len(parts) > 1:
                    package = parts[0]
                else:
                    package = "(default)"
                package_counter[package] += 1
            except Exception:
                continue

        return [
            {"package": pkg, "class_count": count}
            for pkg, count in package_counter.most_common(20)
        ]

    def extract(self):
        if not self._is_valid_apk():
            self.log.error(f"Invalid APK for {self.hash.sha256}")
            return None

        from androguard.core.dex import DEX

        try:
            dex_names = list(self.apk.get_dex_names() or [])
        except Exception as e:
            self.log.warning(f"Error getting DEX names for {self.hash.sha256}: {e}")
            dex_names = []

        try:
            dex_buffers = list(self.apk.get_all_dex() or [])
        except Exception as e:
            self.log.error(f"Error getting DEX buffers for {self.hash.sha256}: {e}")
            return None

        if not dex_buffers:
            self.log.debug("No DEX files found in APK")
            return None

        self.dex_files = []
        for i, dex_data in enumerate(dex_buffers):
            dex_name = dex_names[i] if i < len(dex_names) else f"classes{i}.dex"

            dex_sha256 = hashlib.sha256(dex_data).hexdigest()
            dex_tlsh = None
            if len(dex_data) >= 50:
                try:
                    dex_tlsh = tlsh.hash(dex_data) or None
                except Exception:
                    pass

            try:
                d = DEX(dex_data)
            except Exception as e:
                self.log.warning(f"Failed to parse DEX {dex_name}: {e}")
                continue

            try:
                classes = list(d.get_classes() or [])
            except Exception as e:
                self.log.warning(f"Error getting classes from {dex_name}: {e}")
                classes = []
            try:
                methods = list(d.get_methods() or [])
            except Exception as e:
                self.log.warning(f"Error getting methods from {dex_name}: {e}")
                methods = []
            try:
                strings = list(d.get_strings() or [])
            except Exception as e:
                self.log.warning(f"Error getting strings from {dex_name}: {e}")
                strings = []

            try:
                top_packages = self._compute_top_packages(classes)
            except Exception as e:
                self.log.warning(f"Error computing top packages for {dex_name}: {e}")
                top_packages = []
            try:
                api_usage = self._analyze_api_usage(d)
            except Exception as e:
                self.log.warning(f"Error analyzing API usage for {dex_name}: {e}")
                api_usage = {}
            try:
                obfuscation_indicators = self._compute_obfuscation_indicators(classes, methods)
            except Exception as e:
                self.log.warning(f"Error computing obfuscation indicators for {dex_name}: {e}")
                obfuscation_indicators = None

            self.dex_files.append(APKDexFile(
                filename=dex_name,
                sha256=dex_sha256,
                class_count=len(classes),
                method_count=len(methods),
                string_count=len(strings),
                tlsh=dex_tlsh,
                top_packages=top_packages,
                api_usage=api_usage if api_usage else None,
                obfuscation_indicators=obfuscation_indicators,
            ))

        return self.dex_files if self.dex_files else None

    def prepare_export_data(self, exporter_type: str) -> Any:
        if exporter_type == "ClickHouseExporter":
            if not self.dex_files:
                return None

            current_time = datetime.now(timezone.utc)

            # DEX table: one row per DEX file
            dex_data = []
            for dex in self.dex_files:
                dex_data.append([
                    self.sha256,
                    dex.sha256,
                    dex.filename,
                    dex.tlsh,
                    dex.class_count,
                    dex.method_count,
                    dex.string_count,
                    json.dumps(dex.top_packages) if dex.top_packages else None,
                    json.dumps(dex.obfuscation_indicators) if dex.obfuscation_indicators else None,
                    current_time,
                ])

            # API usage table: one row per category per DEX
            api_data = []
            for dex in self.dex_files:
                if dex.api_usage:
                    for category, api_calls in dex.api_usage.items():
                        api_data.append([
                            self.sha256,
                            dex.sha256,
                            category,
                            api_calls,
                            current_time,
                        ])

            return {
                'multi_table': True,
                'dex': {
                    'table': 'redb_apk_dex',
                    'data': dex_data,
                    'column_names': [
                        'sha256', 'dex_sha256', 'dex_filename', 'dex_tlsh',
                        'dex_class_count', 'dex_method_count', 'dex_string_count',
                        'dex_top_packages', 'dex_obfuscation_indicators',
                        'analysis_date',
                    ],
                    'column_type_names': [
                        'FixedString(64)', 'FixedString(64)', 'String', 'Nullable(String)',
                        'UInt32', 'UInt32', 'UInt32',
                        'Nullable(String)', 'Nullable(String)',
                        "DateTime64(3, 'UTC')",
                    ],
                },
                'api_usage': {
                    'table': 'redb_apk_dex_api_usage',
                    'data': api_data,
                    'column_names': [
                        'sha256', 'dex_sha256', 'api_category',
                        'api_calls', 'analysis_date',
                    ],
                    'column_type_names': [
                        'FixedString(64)', 'FixedString(64)', 'LowCardinality(String)',
                        'Array(String)', "DateTime64(3, 'UTC')",
                    ],
                },
            }

    def get_clickhouse_table(self) -> str:
        return "redb_apk_dex"