Weihao Li

import inspect
import zipfile
from datetime import datetime, timezone
from typing import Any

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


class APKFeaturesExtractor(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.apk_features = None
        self.log.debug(inspect.currentframe().f_code.co_name)

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

    def _extract_zip_timestamps(self):
        """Extract earliest and latest content modification from ZIP entries."""
        earliest = None
        latest = None
        try:
            zf = self._get_zip_file()
            if zf:
                with zf:
                    for info in zf.infolist():
                        try:
                            dt = datetime(*info.date_time)
                            if earliest is None or dt < earliest:
                                earliest = dt
                            if latest is None or dt > latest:
                                latest = dt
                        except (ValueError, TypeError):
                            continue
        except Exception as e:
            self.log.warning(f"Error extracting ZIP timestamps: {e}")
        return (
            earliest.isoformat() if earliest else None,
            latest.isoformat() if latest else None,
        )

    def _extract_supported_abis(self):
        """Determine supported ABIs from lib/ directory."""
        abis = set()
        for f in self._list_files():
            if f.startswith("lib/") and f.endswith(".so"):
                parts = f.split("/")
                if len(parts) >= 3:
                    abis.add(parts[1])
        return sorted(abis)

    def _count_dex_files(self):
        """Count DEX files and compute total size."""
        dex_count = 0
        total_size = 0
        try:
            zf = self._get_zip_file()
            if zf:
                with zf:
                    for info in zf.infolist():
                        if info.filename.endswith(".dex"):
                            dex_count += 1
                            total_size += info.file_size
        except Exception as e:
            self.log.warning(f"Error counting DEX files: {e}")
        return dex_count, total_size

    def _check_embedded_apk(self):
        """Check if the archive contains nested APK files."""
        for f in self._list_files():
            if f.lower().endswith(".apk"):
                return True
        return False

    def _safe_extract(self, field_name, func, default=None):
        """Extract a single field, logging and returning default on failure."""
        try:
            return func()
        except Exception as e:
            self.log.warning(
                f"Error extracting APK field '{field_name}' for "
                f"{self.hash.sha256}: {e}"
            )
            return default

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

        package_name = self._safe_extract(
            "package_name", lambda: self.apk.get_package()
        )
        app_name = self._safe_extract(
            "app_name", lambda: self.apk.get_app_name()
        )

        def _get_version_code():
            try:
                return int(self.apk.get_androidversion_code() or 0)
            except (ValueError, TypeError):
                return 0

        version_code = self._safe_extract("version_code", _get_version_code)
        version_name = self._safe_extract(
            "version_name", lambda: self.apk.get_androidversion_name()
        )

        def _get_sdk(getter):
            val = getter()
            return int(val) if val else None

        min_sdk = self._safe_extract(
            "min_sdk_version", lambda: _get_sdk(self.apk.get_min_sdk_version)
        )
        target_sdk = self._safe_extract(
            "target_sdk_version", lambda: _get_sdk(self.apk.get_target_sdk_version)
        )
        compile_sdk = self._safe_extract(
            "compile_sdk_version",
            lambda: _get_sdk(self.apk.get_effective_target_sdk_version),
        )

        main_activity = self._safe_extract(
            "main_activity", lambda: self.apk.get_main_activity()
        )

        is_debuggable = self._safe_extract(
            "is_debuggable",
            lambda: self.apk.get_attribute_value(
                "application", "debuggable"
            ) == "true",
            False,
        )
        allow_backup = self._safe_extract(
            "allow_backup",
            lambda: self.apk.get_attribute_value(
                "application", "allowBackup"
            ) != "false",
            True,
        )
        uses_cleartext = self._safe_extract(
            "uses_cleartext_traffic",
            lambda: self.apk.get_attribute_value(
                "application", "usesCleartextTraffic"
            ) == "true",
            False,
        )

        supported_abis = self._safe_extract(
            "supported_abis", self._extract_supported_abis, []
        )
        dex_count, total_dex_size = self._safe_extract(
            "dex_files", self._count_dex_files, (0, 0)
        )
        all_files = self._safe_extract("file_list", self._list_files, [])
        total_file_count = len(all_files)
        has_native_code = any(
            f.startswith("lib/") and f.endswith(".so") for f in all_files
        )
        has_assets = any(f.startswith("assets/") for f in all_files)
        contains_embedded_apk = self._safe_extract(
            "contains_embedded_apk", self._check_embedded_apk, False
        )

        uses_libraries = self._safe_extract(
            "uses_libraries",
            lambda: list(self.apk.get_libraries() or []),
            [],
        )

        earliest_mod, latest_mod = self._safe_extract(
            "zip_timestamps", self._extract_zip_timestamps, (None, None)
        )

        self.apk_features = APKFeatures(
            package_name=package_name,
            app_name=app_name,
            version_code=version_code,
            version_name=version_name,
            min_sdk_version=min_sdk,
            target_sdk_version=target_sdk,
            compile_sdk_version=compile_sdk,
            main_activity=main_activity,
            is_debuggable=is_debuggable,
            allow_backup=allow_backup,
            uses_cleartext_traffic=uses_cleartext,
            supported_abis=supported_abis,
            dex_count=dex_count,
            total_dex_size=total_dex_size,
            total_file_count=total_file_count,
            has_native_code=has_native_code,
            has_assets=has_assets,
            uses_libraries=uses_libraries,
            earliest_content_modification=earliest_mod,
            latest_content_modification=latest_mod,
            contains_embedded_apk=contains_embedded_apk,
        )
        return self.apk_features

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

            f = self.apk_features
            current_time = datetime.now(timezone.utc)
            data = [[
                self.sha256,
                f.package_name,
                f.app_name,
                f.version_code,
                f.version_name,
                f.min_sdk_version,
                f.target_sdk_version,
                f.compile_sdk_version,
                f.main_activity,
                int(f.is_debuggable),
                int(f.allow_backup),
                int(f.uses_cleartext_traffic),
                f.supported_abis,
                f.dex_count,
                f.total_dex_size,
                f.total_file_count,
                int(f.has_native_code),
                int(f.has_assets),
                f.uses_libraries,
                f.earliest_content_modification,
                f.latest_content_modification,
                int(f.contains_embedded_apk),
                current_time,
            ]]

            column_names = [
                "sha256",
                "package_name", "app_name", "version_code", "version_name",
                "min_sdk_version", "target_sdk_version", "compile_sdk_version",
                "main_activity",
                "is_debuggable", "allow_backup", "uses_cleartext_traffic",
                "supported_abis",
                "dex_count", "total_dex_size", "total_file_count",
                "has_native_code", "has_assets",
                "uses_libraries",
                "earliest_content_modification", "latest_content_modification",
                "contains_embedded_apk",
                "analysis_date",
            ]

            column_type_names = [
                "FixedString(64)",
                "Nullable(String)", "Nullable(String)", "Nullable(UInt32)", "Nullable(String)",
                "Nullable(UInt16)", "Nullable(UInt16)", "Nullable(UInt16)",
                "Nullable(String)",
                "UInt8", "UInt8", "UInt8",
                "Array(String)",
                "UInt16", "UInt64", "UInt32",
                "UInt8", "UInt8",
                "Array(String)",
                "Nullable(String)", "Nullable(String)",
                "UInt8",
                "DateTime64(3, 'UTC')",
            ]

            return (data, column_names, column_type_names)

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