Xiangyang Huang

import inspect
import json
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 APKManifest, APKManifestComponent


class APKManifestExtractor(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.manifest = None
        self.components = []
        self.log.debug(inspect.currentframe().f_code.co_name)

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

    def _is_component_exported(self, component_type, component_name):
        """Determine if a component is exported.

        Pre-API 31: exported is implicitly True if intent filters exist.
        API 31+: android:exported must be explicit; default is False.
        """
        try:
            exported_attr = None
            # Try to get the exported attribute directly from the XML
            axml = self.apk.get_android_manifest_xml()
            if axml is not None:
                for node in axml.getElementsByTagName(component_type):
                    name = node.getAttributeNS(
                        "http://schemas.android.com/apk/res/android", "name"
                    )
                    if name == component_name:
                        exported_attr = node.getAttributeNS(
                            "http://schemas.android.com/apk/res/android", "exported"
                        )
                        break
        except Exception:
            exported_attr = None

        if exported_attr == "true":
            return True
        if exported_attr == "false":
            return False

        # If not explicitly set, check for intent filters (pre-API 31 behavior)
        try:
            intent_filters = self.apk.get_intent_filters(component_type, component_name)
            if intent_filters:
                actions = intent_filters.get("action", [])
                if actions:
                    return True
        except Exception:
            pass

        return False

    def _get_intent_filters_for_component(self, component_type, component_name):
        """Get intent filters for a specific component."""
        actions = []
        categories = []
        try:
            intent_filters = self.apk.get_intent_filters(component_type, component_name)
            if intent_filters:
                actions = intent_filters.get("action", [])
                categories = intent_filters.get("category", [])
        except Exception:
            pass
        return actions, categories

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

        activities = self._safe_extract(
            "activities", lambda: list(self.apk.get_activities() or []), []
        )
        services = self._safe_extract(
            "services", lambda: list(self.apk.get_services() or []), []
        )
        receivers = self._safe_extract(
            "receivers", lambda: list(self.apk.get_receivers() or []), []
        )
        providers = self._safe_extract(
            "providers", lambda: list(self.apk.get_providers() or []), []
        )

        # Build component list with intent filter info
        self.components = []
        all_actions = set()
        all_categories = set()
        exported_components = []

        component_map = [
            ("activity", activities),
            ("service", services),
            ("receiver", receivers),
            ("provider", providers),
        ]

        for comp_type, comp_list in component_map:
            for comp_name in comp_list:
                try:
                    is_exported = self._is_component_exported(comp_type, comp_name)
                    actions, categories = self._get_intent_filters_for_component(
                        comp_type, comp_name
                    )
                    all_actions.update(actions)
                    all_categories.update(categories)
                    if is_exported:
                        exported_components.append(comp_name)

                    self.components.append(APKManifestComponent(
                        component_type=comp_type,
                        class_name=comp_name,
                        is_exported=is_exported,
                        intent_actions=list(actions),
                        intent_categories=list(categories),
                    ))
                except Exception as e:
                    self.log.warning(
                        f"Error processing component '{comp_name}' for "
                        f"{self.hash.sha256}: {e}"
                    )
                    # Still add the component with minimal info
                    self.components.append(APKManifestComponent(
                        component_type=comp_type,
                        class_name=comp_name,
                        is_exported=False,
                        intent_actions=[],
                        intent_categories=[],
                    ))

        # Uses-feature
        uses_features = []
        try:
            uses_features = list(self.apk.get_features() or [])
        except Exception:
            pass

        # Meta-data
        meta_data = None
        try:
            axml = self.apk.get_android_manifest_xml()
            if axml is not None:
                md = {}
                for node in axml.getElementsByTagName("meta-data"):
                    name = node.getAttributeNS(
                        "http://schemas.android.com/apk/res/android", "name"
                    )
                    value = node.getAttributeNS(
                        "http://schemas.android.com/apk/res/android", "value"
                    )
                    resource = node.getAttributeNS(
                        "http://schemas.android.com/apk/res/android", "resource"
                    )
                    if name:
                        md[name] = value or resource or ""
                if md:
                    meta_data = md
        except Exception:
            pass

        # Full manifest XML
        manifest_xml = None
        try:
            axml = self.apk.get_android_manifest_xml()
            if axml is not None:
                manifest_xml = axml.toxml()
        except Exception:
            try:
                manifest_xml = self.apk.get_android_manifest_axml().get_xml()
                if isinstance(manifest_xml, bytes):
                    manifest_xml = manifest_xml.decode("utf-8", errors="replace")
            except Exception:
                pass

        self.manifest = APKManifest(
            activity_count=len(activities),
            service_count=len(services),
            receiver_count=len(receivers),
            provider_count=len(providers),
            activities=activities,
            services=services,
            receivers=receivers,
            providers=providers,
            exported_components=exported_components,
            intent_filters_by_action=sorted(all_actions),
            intent_filters_by_category=sorted(all_categories),
            uses_features=uses_features,
            meta_data=meta_data,
            manifest_xml=manifest_xml,
        )
        return self.manifest

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

            current_time = datetime.now(timezone.utc)
            m = self.manifest

            # Components table (one row per component)
            components_data = []
            for comp in self.components:
                components_data.append([
                    self.sha256,
                    comp.component_type,
                    comp.class_name,
                    int(comp.is_exported),
                    comp.intent_actions,
                    comp.intent_categories,
                    current_time,
                ])

            # Manifest summary table (one row per APK)
            manifest_data = [[
                self.sha256,
                m.activity_count,
                m.service_count,
                m.receiver_count,
                m.provider_count,
                m.intent_filters_by_action,
                m.intent_filters_by_category,
                m.uses_features,
                m.manifest_xml,
                current_time,
            ]]

            return {
                'multi_table': True,
                'manifest': {
                    'table': 'redb_apk_manifest',
                    'data': manifest_data,
                    'column_names': [
                        'sha256',
                        'activity_count', 'service_count', 'receiver_count', 'provider_count',
                        'intent_filters_by_action', 'intent_filters_by_category',
                        'uses_features', 'manifest_xml', 'analysis_date',
                    ],
                    'column_type_names': [
                        'FixedString(64)',
                        'UInt16', 'UInt16', 'UInt16', 'UInt16',
                        'Array(String)', 'Array(String)',
                        'Array(String)', 'Nullable(String)',
                        "DateTime64(3, 'UTC')",
                    ],
                },
                'components': {
                    'table': 'redb_apk_components',
                    'data': components_data,
                    'column_names': [
                        'sha256', 'component_type', 'class_name', 'is_exported',
                        'intent_actions', 'intent_categories', 'analysis_date',
                    ],
                    'column_type_names': [
                        'FixedString(64)', 'LowCardinality(String)', 'String', 'UInt8',
                        'Array(String)', 'Array(String)',
                        "DateTime64(3, 'UTC')",
                    ],
                },
            }

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