Kanchan Pranay Patil

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"