WanXia Qu

import inspect
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 APKPermissions, APKPermission

# Known dangerous permissions (Android runtime permissions)
DANGEROUS_PERMISSIONS = {
    "android.permission.READ_CALENDAR",
    "android.permission.WRITE_CALENDAR",
    "android.permission.CAMERA",
    "android.permission.READ_CONTACTS",
    "android.permission.WRITE_CONTACTS",
    "android.permission.GET_ACCOUNTS",
    "android.permission.ACCESS_FINE_LOCATION",
    "android.permission.ACCESS_COARSE_LOCATION",
    "android.permission.ACCESS_BACKGROUND_LOCATION",
    "android.permission.RECORD_AUDIO",
    "android.permission.READ_PHONE_STATE",
    "android.permission.READ_PHONE_NUMBERS",
    "android.permission.CALL_PHONE",
    "android.permission.ANSWER_PHONE_CALLS",
    "android.permission.ADD_VOICEMAIL",
    "android.permission.USE_SIP",
    "android.permission.BODY_SENSORS",
    "android.permission.SEND_SMS",
    "android.permission.RECEIVE_SMS",
    "android.permission.READ_SMS",
    "android.permission.RECEIVE_WAP_PUSH",
    "android.permission.RECEIVE_MMS",
    "android.permission.READ_EXTERNAL_STORAGE",
    "android.permission.WRITE_EXTERNAL_STORAGE",
    "android.permission.READ_MEDIA_IMAGES",
    "android.permission.READ_MEDIA_VIDEO",
    "android.permission.READ_MEDIA_AUDIO",
    "android.permission.POST_NOTIFICATIONS",
    "android.permission.NEARBY_WIFI_DEVICES",
    "android.permission.ACTIVITY_RECOGNITION",
    "android.permission.BLUETOOTH_SCAN",
    "android.permission.BLUETOOTH_ADVERTISE",
    "android.permission.BLUETOOTH_CONNECT",
}


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

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

    def _get_protection_level(self, perm_name):
        """Determine the protection level of a permission.

        Uses androguard's permission mappings first, falls back to the
        known dangerous list.
        """
        try:
            from androguard.core.api_specific_resources import load_permission_mappings
            target_sdk = self.apk.get_target_sdk_version()
            sdk = int(target_sdk) if target_sdk else 30
            permission_map = load_permission_mappings(sdk)
            if perm_name in permission_map:
                info = permission_map[perm_name]
                if isinstance(info, dict):
                    return info.get("protectionLevel", "unknown")
                return "unknown"
        except Exception:
            pass

        # Fallback to known dangerous set
        if perm_name in DANGEROUS_PERMISSIONS:
            return "dangerous"
        if perm_name.startswith("android.permission."):
            return "normal"
        return "unknown"

    def _compute_permhash(self):
        """Compute permhash using the permhash library."""
        try:
            from permhash.functions import permhash_apk
            return permhash_apk(self.filepath)
        except Exception as e:
            self.log.warning(f"Failed to compute permhash: {e}")
            return None

    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 permissions 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

        requested_perms = self._safe_extract(
            "requested_permissions",
            lambda: list(self.apk.get_permissions() or []), []
        )
        custom_perms = self._safe_extract(
            "custom_permissions",
            lambda: list(self.apk.get_declared_permissions() or []), []
        )
        custom_perm_set = set(custom_perms)

        permission_details = []
        dangerous_permissions = []

        for perm in requested_perms:
            try:
                protection_level = self._get_protection_level(perm)
                is_custom = perm in custom_perm_set

                permission_details.append(APKPermission(
                    permission_name=perm,
                    protection_level=protection_level,
                    is_custom=is_custom,
                ))

                if protection_level == "dangerous":
                    dangerous_permissions.append(perm)
            except Exception as e:
                self.log.warning(
                    f"Error processing permission '{perm}' for "
                    f"{self.hash.sha256}: {e}"
                )
                permission_details.append(APKPermission(
                    permission_name=perm,
                    protection_level="unknown",
                    is_custom=perm in custom_perm_set,
                ))

        permhash_value = self._compute_permhash()

        self.permissions_data = APKPermissions(
            total_permission_count=len(requested_perms),
            dangerous_permission_count=len(dangerous_permissions),
            permissions=requested_perms,
            dangerous_permissions=dangerous_permissions,
            custom_permissions=custom_perms,
            permission_details=permission_details,
            permhash=permhash_value,
        )
        return self.permissions_data

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

            current_time = datetime.now(timezone.utc)

            # One row per permission
            data = []
            for detail in self.permissions_data.permission_details:
                data.append([
                    self.sha256,
                    detail.permission_name,
                    detail.protection_level,
                    int(detail.is_custom),
                    current_time,
                ])

            if not data:
                return None

            column_names = [
                'sha256', 'permission_name', 'protection_level',
                'is_custom', 'analysis_date',
            ]

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

            return (data, column_names, column_type_names)

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