Xiaohu Fan

import inspect
import re
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 APKInconsistencyTests

# Emulator detection indicator strings
EMULATOR_INDICATORS = {
    "generic", "sdk", "google_sdk", "Emulator",
    "goldfish", "ranchu", "Andy", "Genymotion",
    "BlueStacks", "nox", "ttVM_Hdragon",
}

# Root detection indicator strings
ROOT_INDICATORS = {
    "/system/app/Superuser.apk",
    "/system/xbin/su",
    "/system/bin/su",
    "com.noshufou.android.su",
    "com.thirdparty.superuser",
    "eu.chainfire.supersu",
    "com.koushikdutta.superuser",
    "com.topjohnwu.magisk",
}

# Standard DEX filename pattern
STANDARD_DEX_PATTERN = re.compile(r"^classes\d*\.dex$")


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

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

    def _test_zip_bomb(self):
        """Check if any ZIP entry has compression ratio > 100:1."""
        try:
            zf = self._get_zip_file()
            if not zf:
                return None
            with zf:
                for info in zf.infolist():
                    if info.compress_size > 0:
                        ratio = info.file_size / info.compress_size
                        if ratio > 100:
                            return True
            return False
        except Exception as e:
            self.log.warning(f"Error in zip bomb test: {e}")
            return None

    def _test_zip_duplicate_entries(self):
        """Check for duplicate filenames in ZIP directory."""
        try:
            zf = self._get_zip_file()
            if not zf:
                return None
            with zf:
                names = [info.filename for info in zf.infolist()]
                return len(names) != len(set(names))
        except Exception as e:
            self.log.warning(f"Error in duplicate entries test: {e}")
            return None

    def _test_zip_path_traversal(self):
        """Check for path traversal (../) in ZIP entry names."""
        try:
            for f in self._list_files():
                if ".." in f or f.startswith("/"):
                    return True
            return False
        except Exception as e:
            self.log.warning(f"Error in path traversal test: {e}")
            return None

    def _test_zip_suspicious_timestamps(self):
        """Check for timestamps at epoch (1980) or in the future."""
        try:
            zf = self._get_zip_file()
            if not zf:
                return None
            now = datetime.now()
            with zf:
                for info in zf.infolist():
                    try:
                        dt = datetime(*info.date_time)
                        if dt.year <= 1980 or dt > now:
                            return True
                    except (ValueError, TypeError):
                        continue
            return False
        except Exception as e:
            self.log.warning(f"Error in suspicious timestamps test: {e}")
            return None

    def _test_hidden_dex_files(self):
        """Check for DEX files not matching classes*.dex pattern."""
        try:
            for f in self._list_files():
                if f.endswith(".dex"):
                    basename = f.split("/")[-1]
                    if not STANDARD_DEX_PATTERN.match(basename):
                        return True
            return False
        except Exception as e:
            self.log.warning(f"Error in hidden DEX files test: {e}")
            return None

    def _test_manifest_component_mismatch(self):
        """Check for declared components that don't exist in DEX."""
        try:
            if not self._is_valid_apk():
                return None

            # Get all class names from DEX
            dex_classes = set()
            try:
                from androguard.core.dex import DEX
                for dex_data in (self.apk.get_all_dex() or []):
                    try:
                        d = DEX(dex_data)
                        for cls in d.get_classes():
                            name = cls.get_name()
                            if name:
                                # Convert "Lcom/example/Foo;" to "com.example.Foo"
                                dex_classes.add(
                                    name[1:-1].replace("/", ".") if name.startswith("L") else name
                                )
                    except Exception:
                        continue
            except Exception:
                return None

            if not dex_classes:
                return None

            # Check manifest components against DEX classes
            components = []
            try:
                components.extend(self.apk.get_activities() or [])
                components.extend(self.apk.get_services() or [])
                components.extend(self.apk.get_receivers() or [])
                components.extend(self.apk.get_providers() or [])
            except Exception:
                return None

            for comp in components:
                if comp and comp not in dex_classes:
                    # Component might use a shorthand; check with package prefix
                    package = self.apk.get_package() or ""
                    full_name = package + comp if comp.startswith(".") else comp
                    if full_name not in dex_classes:
                        return True

            return False
        except Exception as e:
            self.log.warning(f"Error in manifest component mismatch test: {e}")
            return None

    def _test_debuggable_release(self):
        """Check android:debuggable=true combined with a release signature."""
        try:
            if not self._is_valid_apk():
                return None

            is_debuggable = self.apk.get_attribute_value(
                "application", "debuggable"
            ) == "true"

            if not is_debuggable:
                return False

            # Check if it has a signing certificate (release builds have certs)
            try:
                certs = self.apk.get_certificates()
                if certs and len(certs) > 0:
                    return True
            except Exception:
                pass

            return False
        except Exception as e:
            self.log.warning(f"Error in debuggable release test: {e}")
            return None

    def _get_dex_strings(self):
        """Get all string constants from DEX files."""
        all_strings = set()
        try:
            from androguard.core.dex import DEX
            for dex_data in (self.apk.get_all_dex() or []):
                try:
                    d = DEX(dex_data)
                    for s in d.get_strings():
                        if s:
                            all_strings.add(s)
                except Exception:
                    continue
        except Exception:
            pass
        return all_strings

    def _test_emulator_detection_strings(self):
        """Check for emulator detection patterns in DEX strings."""
        try:
            if not self._is_valid_apk():
                return None

            dex_strings = self._get_dex_strings()
            for indicator in EMULATOR_INDICATORS:
                for s in dex_strings:
                    if indicator in s:
                        return True
            return False
        except Exception as e:
            self.log.warning(f"Error in emulator detection test: {e}")
            return None

    def _test_debugger_detection(self):
        """Check for debugger detection API calls in DEX."""
        try:
            if not self._is_valid_apk():
                return None

            dex_strings = self._get_dex_strings()
            debugger_patterns = {
                "isDebuggerConnected",
                "waitingForDebugger",
                "Debug.isDebuggerConnected",
            }
            for pattern in debugger_patterns:
                for s in dex_strings:
                    if pattern in s:
                        return True
            return False
        except Exception as e:
            self.log.warning(f"Error in debugger detection test: {e}")
            return None

    def _test_root_detection(self):
        """Check for root detection patterns in DEX."""
        try:
            if not self._is_valid_apk():
                return None

            dex_strings = self._get_dex_strings()
            for indicator in ROOT_INDICATORS:
                for s in dex_strings:
                    if indicator in s:
                        return True
            return False
        except Exception as e:
            self.log.warning(f"Error in root detection test: {e}")
            return None

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

        self.test_results = APKInconsistencyTests(
            test_zip_bomb=self._test_zip_bomb(),
            test_zip_duplicate_entries=self._test_zip_duplicate_entries(),
            test_zip_path_traversal=self._test_zip_path_traversal(),
            test_zip_suspicious_timestamps=self._test_zip_suspicious_timestamps(),
            test_hidden_dex_files=self._test_hidden_dex_files(),
            test_manifest_component_mismatch=self._test_manifest_component_mismatch(),
            test_debuggable_release=self._test_debuggable_release(),
            test_emulator_detection_strings=self._test_emulator_detection_strings(),
            test_debugger_detection=self._test_debugger_detection(),
            test_root_detection=self._test_root_detection(),
        )
        return self.test_results

    def _bool_to_nullable(self, val):
        """Convert bool/None to ClickHouse Nullable(UInt8)."""
        if val is None:
            return None
        return int(val)

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

            current_time = datetime.now(timezone.utc)
            t = self.test_results

            data = [[
                self.sha256,
                self._bool_to_nullable(t.test_zip_bomb),
                self._bool_to_nullable(t.test_zip_duplicate_entries),
                self._bool_to_nullable(t.test_zip_path_traversal),
                self._bool_to_nullable(t.test_zip_suspicious_timestamps),
                self._bool_to_nullable(t.test_hidden_dex_files),
                self._bool_to_nullable(t.test_manifest_component_mismatch),
                self._bool_to_nullable(t.test_debuggable_release),
                self._bool_to_nullable(t.test_emulator_detection_strings),
                self._bool_to_nullable(t.test_debugger_detection),
                self._bool_to_nullable(t.test_root_detection),
                current_time,
            ]]

            column_names = [
                'sha256',
                'test_zip_bomb', 'test_zip_duplicate_entries',
                'test_zip_path_traversal', 'test_zip_suspicious_timestamps',
                'test_hidden_dex_files', 'test_manifest_component_mismatch',
                'test_debuggable_release', 'test_emulator_detection_strings',
                'test_debugger_detection', 'test_root_detection',
                'analysis_date',
            ]

            column_type_names = [
                'FixedString(64)',
                'Nullable(UInt8)', 'Nullable(UInt8)',
                'Nullable(UInt8)', 'Nullable(UInt8)',
                'Nullable(UInt8)', 'Nullable(UInt8)',
                'Nullable(UInt8)', 'Nullable(UInt8)',
                'Nullable(UInt8)', 'Nullable(UInt8)',
                "DateTime64(3, 'UTC')",
            ]

            return (data, column_names, column_type_names)

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