Hanlin Gu

import hashlib
import inspect
import os
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 APKResource


# ─── Suspicious file types ──────────────────────────────────────────────
# File types that are suspicious when found inside res/ or assets/.
# Excludes javascript/html (extremely common in legitimate hybrid apps)
# and common media/font types that are normal APK content.
SUSPICIOUS_TYPES = {
    # Executables — no legitimate reason in assets/res
    "elf", "pebin", "macho", "dex", "apk",
    # Java containers — DexClassLoader target
    "jar",
    # Archives — rare in legitimate assets (~135:1 malware-to-benign ratio)
    "zip", "gzip", "7z", "xz", "tar", "bzip2", "rar", "7zip", "lzma",
    # Scripts with system execution capability
    "shell", "python", "powershell", "batch",
}

# ─── Entropy thresholds ─────────────────────────────────────────────────
# For unrecognized/unknown types: encrypted payloads typically land > 7.0
ENTROPY_HIGH_UNKNOWN = 7.0
# For recognized-but-non-image types: stricter threshold
ENTROPY_EXTREME = 7.85

# ─── Android-specific binary format magic bytes ─────────────────────────
# These formats are common in legitimate APKs but unknown to Magika,
# causing misclassification (e.g., AXML → "gzip", profm → "unknown").
AXML_MAGIC = b'\x03\x00\x08\x00'       # Android Binary XML (compiled res/*.xml)
ARSC_MAGIC = b'\x02\x00\x0c\x00'       # Android compiled resource table
ART_PROF_MAGIC = b'pro\x00'            # ART baseline profile
ART_PROFM_MAGIC = b'prm\x00'           # ART baseline profile metadata

# ─── Allowlisted paths ──────────────────────────────────────────────────
# Fixed, hardcoded paths in the Android build system that are always benign.
# ART profiles at these exact paths are shipped by Jetpack ProfileInstaller.
ALLOWLISTED_PATHS = {
    "assets/dexopt/baseline.prof",
    "assets/dexopt/baseline.profm",
}

# ─── Image handling ─────────────────────────────────────────────────────
# Magika-confirmed image types: high entropy is expected (lossy codecs
# like VP8/JPEG arithmetic-code toward entropy ~7.95-8.0 by design).
IMAGE_MAGIKA_TYPES = {"png", "webp", "jpeg", "gif", "bmp", "tiff", "ico"}
IMAGE_EXTENSIONS = {".png", ".webp", ".jpg", ".jpeg", ".gif", ".bmp", ".tiff", ".ico"}

# ─── Types Magika assigns when it can't identify the content ────────────
UNRECOGNIZED_MAGIKA_TYPES = {"unknown", "empty"}

# ─── Resource scan limits ───────────────────────────────────────────────
MAX_RESOURCE_FILES = 5000


class APKResourceExtractor(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.resources = []
        self.suspicious_files = []
        self.log.debug(inspect.currentframe().f_code.co_name)

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

    # ─── Core classification logic ──────────────────────────────────────

    def _identify_android_format(self, header: bytes) -> str | None:
        """
        Identify Android-specific binary formats that Magika doesn't know.
        Returns a corrected type label, or None to fall through to Magika.
        """
        if len(header) < 4:
            return None

        magic4 = header[:4]

        # Android Binary XML — all res/*.xml in a compiled APK.
        # Magika often misclassifies this as "gzip".
        if magic4 == AXML_MAGIC:
            return "android_binary_xml"

        # Android compiled resource table (resources.arsc chunks)
        if magic4 == ARSC_MAGIC:
            return "android_resource_table"

        # ART baseline profiles — high entropy (zlib inside) but benign.
        # The format is inert (method reference bitmaps/metadata, not
        # executable code) and some build configs place them at varying paths.
        if magic4 == ART_PROF_MAGIC:
            return "android_art_profile"
        if magic4 == ART_PROFM_MAGIC:
            return "android_art_profile_metadata"

        return None

    def _is_suspicious_resource(
        self, path: str, magika_type: str, entropy: float,
        android_type: str | None,
    ) -> bool:
        """
        Determine if a resource file is suspicious.

        Detection layers:
        1. Allowlisted paths → always benign
        2. Android-specific format override → reclassify Magika mislabels
        3. Image extension vs Magika type mismatch → encrypted blob detection
        4. Magika-confirmed images → benign regardless of entropy
        5. Suspicious type match → flag known-dangerous types
        6. High-entropy unknown blobs → likely encrypted payloads
        """

        # ── Layer 1: Allowlisted paths (hardcoded Android build artifacts) ──
        if path in ALLOWLISTED_PATHS:
            return False

        # ── Layer 2: Android-specific format detection ──────────────────────
        # Override Magika's label for formats it doesn't recognize.
        # All Android-specific formats (AXML, ARSC, ART profiles) are
        # legitimate build artifacts — never suspicious.
        if android_type is not None:
            return False

        # ── Layer 3: Image extension / Magika type mismatch ─────────────────
        # If the file extension claims "image" but Magika's content analysis
        # disagrees, this is a strong signal for an encrypted payload with
        # a fake image extension (e.g., ErrorFather's "rbyypivsnw.png").
        ext = os.path.splitext(path)[1].lower()
        if ext in IMAGE_EXTENSIONS and magika_type not in IMAGE_MAGIKA_TYPES:
            # Exception: Magika might label a valid image as "unknown" if
            # the file is very small (< ~16 bytes). Don't flag tiny files.
            if entropy > 5.0:
                return True

        # ── Layer 4: Magika-confirmed images → benign ───────────────────────
        # Lossy codecs (VP8, JPEG) produce entropy up to ~8.0 by design.
        # If Magika confirms image structure, high entropy is expected.
        if magika_type in IMAGE_MAGIKA_TYPES:
            return False

        # ── Layer 5: Known suspicious file types ────────────────────────────
        if magika_type in SUSPICIOUS_TYPES:
            return True

        # ── Layer 6: High-entropy unrecognized blobs ────────────────────────
        # Files Magika can't identify with high entropy are likely encrypted
        # payloads. Most Android malware packers store encrypted DEX/SO
        # payloads as opaque blobs with random names and no valid magic.
        if magika_type in UNRECOGNIZED_MAGIKA_TYPES and entropy > ENTROPY_HIGH_UNKNOWN:
            return True

        # ── Layer 7: Extreme entropy on any non-image recognized type ───────
        # Catches edge cases where Magika assigns a benign label (e.g.,
        # "xml", "txt") but the entropy is impossibly high for that format.
        if magika_type not in IMAGE_MAGIKA_TYPES and entropy > ENTROPY_EXTREME:
            return True

        return False

    # ─── Extraction pipeline ────────────────────────────────────────────

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

        try:
            from magika import Magika
            magika = Magika()
        except Exception as e:
            self.log.error(f"Failed to initialize Magika for {self.hash.sha256}: {e}")
            magika = None

        self.resources = []
        self.suspicious_files = []
        scanned = 0

        zf = self._get_zip_file()
        if not zf:
            return None

        with zf:
            for info in zf.infolist():
                if info.is_dir():
                    continue
                if not (info.filename.startswith("res/") or
                        info.filename.startswith("assets/")):
                    continue

                if scanned >= MAX_RESOURCE_FILES:
                    self.log.warning(
                        f"Resource scan limit reached ({MAX_RESOURCE_FILES}), "
                        f"stopping resource enumeration"
                    )
                    break
                scanned += 1

                try:
                    data = zf.read(info.filename)
                except Exception as e:
                    self.log.warning(
                        f"Error reading resource {info.filename}: {e}"
                    )
                    continue

                try:
                    file_sha256 = hashlib.sha256(data).hexdigest()
                    file_entropy = round(self.calculate_entropy(data), 3)

                    # Read first bytes for Android-specific format detection
                    header = data[:16] if len(data) >= 16 else data

                    if magika:
                        try:
                            filetype = magika.identify_bytes(data).output.label
                        except Exception:
                            filetype = "unknown"
                    else:
                        filetype = "unknown"

                    # Identify Android-specific formats once, reuse for
                    # both stored type and suspicion classification
                    android_type = self._identify_android_format(header)
                    stored_type = android_type if android_type else filetype

                    suspicious = self._is_suspicious_resource(
                        path=info.filename,
                        magika_type=filetype,
                        entropy=file_entropy,
                        android_type=android_type,
                    )

                    resource = APKResource(
                        path=info.filename,
                        size=info.file_size,
                        sha256=file_sha256,
                        filetype_magika=stored_type,
                        entropy=file_entropy,
                    )

                    if suspicious:
                        resource.is_suspicious = True
                        self.suspicious_files.append(resource)

                    self.resources.append(resource)
                except Exception as e:
                    self.log.warning(
                        f"Error processing resource {info.filename}: {e}"
                    )
                    continue

        if not self.resources:
            return None

        return {
            "total_resource_count": len(self.resources),
            "total_resource_size": sum(r.size for r in self.resources),
            "suspicious_file_count": len(self.suspicious_files),
            "resources": self.resources,
            "suspicious_files": self.suspicious_files,
        }

    # ─── Export ──────────────────────────────────────────────────────────

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

            current_time = datetime.now(timezone.utc)
            data = []
            for res in self.resources:
                data.append([
                    self.sha256,
                    res.path,
                    res.size,
                    res.sha256,
                    res.filetype_magika,
                    res.entropy,
                    int(res.is_suspicious),
                    current_time,
                ])

            column_names = [
                'sha256', 'resource_path', 'resource_size',
                'resource_sha256', 'resource_magika', 'resource_entropy',
                'is_suspicious', 'analysis_date',
            ]

            column_type_names = [
                'FixedString(64)', 'String', 'UInt64',
                'FixedString(64)', 'LowCardinality(String)', 'Float32',
                'UInt8', "DateTime64(3, 'UTC')",
            ]

            return (data, column_names, column_type_names)

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