Weikun Li

"""Tests for APK string extraction and IOC extraction integration.

Covers:
- APKCodeAnalyzer._extract_strings() — string extraction from DEX objects
- APKCodeAnalyzer._string_entropy() — Shannon entropy computation
- DecompileAPK strings export table schema
- IOCExtractorFromResults with APK-format analysis_results
- Worker IOC wiring for APK
"""
import os
import tempfile
from dataclasses import asdict
from unittest.mock import MagicMock, patch

import pytest

from redb.extractors.ioc_extractor.ioc_extractor import IOCExtractorFromResults
from redb.extractors.ioc_extractor.standalone_ioc_extractor import (
    IOCType,
    SourceType,
)

pytestmark = [pytest.mark.unit, pytest.mark.apk]


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _make_analyzer(min_instructions=1):
    """Create an APKCodeAnalyzer with mocked dependencies."""
    with patch(
        "redb.extractors.decompiler.apk.analyzer.JADXDecompiler"
    ), patch(
        "redb.extractors.decompiler.apk.analyzer.ApktoolDisassembler"
    ):
        from redb.extractors.decompiler.apk.analyzer import APKCodeAnalyzer
        analyzer = APKCodeAnalyzer("/fake/path.apk")
        analyzer.min_instructions = min_instructions
        return analyzer


def _make_mock_dex(strings):
    """Create a mock DEX object that returns the given strings."""
    dex = MagicMock()
    dex.get_strings.return_value = strings
    return dex


def _create_fake_apk_file():
    """Create a temporary file to act as a fake APK."""
    fd, path = tempfile.mkstemp(suffix=".apk")
    os.write(fd, b"PK\x03\x04fake apk content for hashing")
    os.close(fd)
    return path


# ---------------------------------------------------------------------------
# Tests: _string_entropy
# ---------------------------------------------------------------------------

class TestStringEntropy:
    """Tests for Shannon entropy computation."""

    def test_empty_string(self):
        analyzer = _make_analyzer()
        assert analyzer._string_entropy("") == 0.0

    def test_single_char(self):
        analyzer = _make_analyzer()
        assert analyzer._string_entropy("a") == 0.0

    def test_repeated_char(self):
        analyzer = _make_analyzer()
        assert analyzer._string_entropy("aaaa") == 0.0

    def test_two_equal_chars(self):
        analyzer = _make_analyzer()
        # "ab" -> entropy = 1.0
        assert abs(analyzer._string_entropy("ab") - 1.0) < 0.01

    def test_higher_entropy(self):
        analyzer = _make_analyzer()
        # More variety = higher entropy
        low = analyzer._string_entropy("aabb")
        high = analyzer._string_entropy("abcd")
        assert high > low

    def test_url_string_entropy(self):
        analyzer = _make_analyzer()
        entropy = analyzer._string_entropy("https://evil.com/payload")
        assert entropy > 2.0  # URLs have moderate entropy


# ---------------------------------------------------------------------------
# Tests: _extract_strings
# ---------------------------------------------------------------------------

class TestExtractStrings:
    """Tests for DEX string extraction."""

    def test_basic_extraction(self):
        analyzer = _make_analyzer()
        dex = _make_mock_dex(["hello", "world"])
        result = analyzer._extract_strings([dex])

        assert len(result) == 2
        assert result[0]["string"] == "hello"
        assert result[1]["string"] == "world"

    def test_string_fields_present(self):
        analyzer = _make_analyzer()
        dex = _make_mock_dex(["test_string"])
        result = analyzer._extract_strings([dex])

        entry = result[0]
        assert "string" in entry
        assert "string_encoding" in entry
        assert "string_offset" in entry
        assert "string_length" in entry
        assert "string_entropy" in entry
        assert entry["string_encoding"] == "UTF8"
        assert entry["string_length"] == len("test_string")

    def test_deduplication_across_dex(self):
        """Same string in multiple DEX files is only extracted once."""
        analyzer = _make_analyzer()
        dex1 = _make_mock_dex(["shared", "unique1"])
        dex2 = _make_mock_dex(["shared", "unique2"])
        result = analyzer._extract_strings([dex1, dex2])

        string_values = [s["string"] for s in result]
        assert string_values.count("shared") == 1
        assert "unique1" in string_values
        assert "unique2" in string_values
        assert len(result) == 3

    def test_empty_strings_filtered(self):
        analyzer = _make_analyzer()
        dex = _make_mock_dex(["", "valid", None, "also_valid"])
        result = analyzer._extract_strings([dex])

        string_values = [s["string"] for s in result]
        assert "" not in string_values
        assert None not in string_values
        assert "valid" in string_values
        assert "also_valid" in string_values

    def test_no_dex_files(self):
        analyzer = _make_analyzer()
        result = analyzer._extract_strings([])
        assert result == []

    def test_dex_get_strings_returns_none(self):
        analyzer = _make_analyzer()
        dex = MagicMock()
        dex.get_strings.return_value = None
        result = analyzer._extract_strings([dex])
        assert result == []

    def test_dex_get_strings_exception(self):
        analyzer = _make_analyzer()
        dex = MagicMock()
        dex.get_strings.side_effect = RuntimeError("corrupt DEX")
        result = analyzer._extract_strings([dex])
        assert result == []

    def test_entropy_computed(self):
        analyzer = _make_analyzer()
        dex = _make_mock_dex(["abcdefgh"])
        result = analyzer._extract_strings([dex])
        assert result[0]["string_entropy"] > 0.0

    def test_incremental_offsets(self):
        """String offsets are incremented sequentially."""
        analyzer = _make_analyzer()
        dex = _make_mock_dex(["a", "b", "c"])
        result = analyzer._extract_strings([dex])

        offsets = [s["string_offset"] for s in result]
        assert offsets == [0, 1, 2]

    def test_url_strings_preserved(self):
        """URL strings are extracted without modification."""
        analyzer = _make_analyzer()
        dex = _make_mock_dex([
            "https://evil.com/payload",
            "http://c2.malware.org/gate",
        ])
        result = analyzer._extract_strings([dex])
        strings = [s["string"] for s in result]
        assert "https://evil.com/payload" in strings
        assert "http://c2.malware.org/gate" in strings


# ---------------------------------------------------------------------------
# Tests: strings in extract() integration
# ---------------------------------------------------------------------------

class TestExtractStringsIntegration:
    """Tests that strings are included in extract() results."""

    @patch("redb.extractors.decompiler.apk.analyzer.APKCodeAnalyzer._run_androguard")
    @patch("redb.extractors.decompiler.apk.analyzer.ApktoolDisassembler")
    @patch("redb.extractors.decompiler.apk.analyzer.JADXDecompiler")
    def test_extract_includes_strings(self, mock_jadx, mock_apktool, mock_androguard):
        from redb.extractors.decompiler.apk.analyzer import APKCodeAnalyzer

        mock_dex = _make_mock_dex(["hello", "world"])
        mock_analysis = MagicMock()
        mock_analysis.get_methods.return_value = []
        mock_androguard.return_value = (MagicMock(), [mock_dex], mock_analysis)

        analyzer = APKCodeAnalyzer("/fake/path.apk", log=MagicMock())
        results = analyzer.extract()

        assert "strings" in results
        assert len(results["strings"]) == 2
        assert results["strings"][0]["string"] == "hello"
        analyzer.cleanup()

    @patch("redb.extractors.decompiler.apk.analyzer.APKCodeAnalyzer._run_androguard")
    @patch("redb.extractors.decompiler.apk.analyzer.ApktoolDisassembler")
    @patch("redb.extractors.decompiler.apk.analyzer.JADXDecompiler")
    def test_extract_strings_error_is_non_fatal(self, mock_jadx, mock_apktool, mock_androguard):
        from redb.extractors.decompiler.apk.analyzer import APKCodeAnalyzer

        # Return a bad dex list that will cause _extract_strings to fail
        mock_analysis = MagicMock()
        mock_analysis.get_methods.return_value = []
        # dexs is not iterable — triggers the except block
        mock_androguard.return_value = (MagicMock(), 42, mock_analysis)

        analyzer = APKCodeAnalyzer("/fake/path.apk", log=MagicMock())
        results = analyzer.extract()

        # Should not crash, strings should be empty, error recorded
        assert results["strings"] == []
        assert any(
            e.get("error_location") == "strings"
            for e in results["analysis_errors"]
        )
        analyzer.cleanup()


# ---------------------------------------------------------------------------
# Tests: DecompileAPK strings export table
# ---------------------------------------------------------------------------

class TestDecompileAPKStringsExport:
    """Tests for strings table in DecompileAPK.prepare_export_data()."""

    def _make_extractor_with_strings(self, strings_data):
        from redb.extractors.decompiler.DecompileAPK import DecompileAPK
        apk_path = _create_fake_apk_file()
        ext = DecompileAPK(apk_path, MagicMock())
        ext.analysis_results = {
            "sha256": "a" * 64,
            "sha1": "b" * 40,
            "md5": "c" * 32,
            "decompiled_content": [],
            "decompiled_refs": [],
            "smali_content": [],
            "smali_refs": [],
            "similarity_metrics": [],
            "strings": strings_data,
            "analysis_errors": [],
        }
        return ext, apk_path

    def test_strings_table_present(self):
        ext, path = self._make_extractor_with_strings([
            {"string": "hello", "string_encoding": "UTF8",
             "string_offset": 0, "string_length": 5, "string_entropy": 2.32}
        ])
        try:
            export = ext.prepare_export_data("ClickHouseExporter")
            assert "strings_raw" in export
        finally:
            os.unlink(path)

    def test_strings_table_schema(self):
        ext, path = self._make_extractor_with_strings([
            {"string": "hello", "string_encoding": "UTF8",
             "string_offset": 0, "string_length": 5, "string_entropy": 2.32}
        ])
        try:
            export = ext.prepare_export_data("ClickHouseExporter")
            sr = export["strings_raw"]
            assert sr["table"] == "code_binja_strings_raw"
            assert len(sr["column_names"]) == 8
            assert len(sr["column_type_names"]) == 8
            assert "sha256" in sr["column_names"]
            assert "string" in sr["column_names"]
            assert "string_raw" in sr["column_names"]
            assert "string_encoding" in sr["column_names"]
            assert "string_offset" in sr["column_names"]
            assert "string_length" in sr["column_names"]
            assert "string_raw_length" in sr["column_names"]
            assert "string_entropy" in sr["column_names"]
            assert len(sr["data"]) == 1
            assert len(sr["data"][0]) == 8
        finally:
            os.unlink(path)

    def test_strings_sha256_propagated(self):
        ext, path = self._make_extractor_with_strings([
            {"string": "test", "string_encoding": "UTF8",
             "string_offset": 0, "string_length": 4, "string_entropy": 2.0}
        ])
        try:
            export = ext.prepare_export_data("ClickHouseExporter")
            row = export["strings_raw"]["data"][0]
            assert row[0] == "a" * 64  # sha256 is first column
        finally:
            os.unlink(path)

    def test_empty_strings_not_exported(self):
        ext, path = self._make_extractor_with_strings([])
        try:
            export = ext.prepare_export_data("ClickHouseExporter")
            assert "strings_raw" not in export
        finally:
            os.unlink(path)


# ---------------------------------------------------------------------------
# Tests: IOCExtractorFromResults with APK format
# ---------------------------------------------------------------------------

class TestIOCExtractorAPKFormat:
    """Tests that IOCExtractorFromResults handles APK analysis_results."""

    def setup_method(self):
        self.log = MagicMock()

    def test_extract_iocs_from_apk_strings(self):
        """IOCs extracted from APK strings (same format as Binja)."""
        results = {
            "strings": [
                {"string": "https://evil.com/payload", "string_offset": 0},
                {"string": "Contact [email protected]", "string_offset": 1},
            ],
            "decompiled": [],
            "decompiled_content": [],
        }
        extractor = IOCExtractorFromResults(results, sha256="a" * 64, log=self.log)
        iocs = extractor.extract()

        url_iocs = [i for i in iocs if i.ioc_type == IOCType.URL]
        assert len(url_iocs) >= 1
        assert any("evil.com" in i.ioc_value for i in url_iocs)

    def test_extract_iocs_from_apk_decompiled_content(self):
        """IOCs extracted from APK's decompiled_content key."""
        results = {
            "strings": [],
            "decompiled": [],  # Binja format — empty
            "decompiled_content": [  # APK format
                {
                    "decompiled_method": "connect('https://c2.malware.org/gate');",
                    "decompiled_method_hash": "hash123",
                    "method_type": "USER",
                }
            ],
        }
        extractor = IOCExtractorFromResults(results, sha256="a" * 64, log=self.log)
        iocs = extractor.extract()

        url_iocs = [i for i in iocs if i.ioc_type == IOCType.URL]
        assert len(url_iocs) >= 1

    def test_apk_library_methods_skipped(self):
        """LIBRARY methods in decompiled_content are skipped."""
        results = {
            "strings": [],
            "decompiled": [],
            "decompiled_content": [
                {
                    "decompiled_method": "https://should-skip.com",
                    "decompiled_method_hash": "lib_hash",
                    "method_type": "LIBRARY",
                }
            ],
        }
        extractor = IOCExtractorFromResults(results, sha256="a" * 64, log=self.log)
        iocs = extractor.extract()
        assert len(iocs) == 0

    def test_combined_binja_and_apk_sources(self):
        """Both Binja and APK format decompiled entries are processed."""
        results = {
            "strings": [],
            "decompiled": [
                {
                    "decompiled_function": "call('https://binja.example.com');",
                    "decompiled_function_hash": "binja_hash",
                    "function_type": "USER",
                }
            ],
            "decompiled_content": [
                {
                    "decompiled_method": "send('https://apk.example.com');",
                    "decompiled_method_hash": "apk_hash",
                    "method_type": "USER",
                }
            ],
        }
        extractor = IOCExtractorFromResults(results, sha256="a" * 64, log=self.log)
        iocs = extractor.extract()

        url_iocs = [i for i in iocs if i.ioc_type == IOCType.URL]
        urls = [i.ioc_value for i in url_iocs]
        assert any("binja.example.com" in u for u in urls)
        assert any("apk.example.com" in u for u in urls)

    def test_apk_results_only(self):
        """Works when only APK keys are present (no 'decompiled' key)."""
        results = {
            "strings": [
                {"string": "https://evil.com", "string_offset": 0},
            ],
            "decompiled_content": [
                {
                    "decompiled_method": "x = 1",
                    "decompiled_method_hash": "h",
                    "method_type": "USER",
                }
            ],
        }
        extractor = IOCExtractorFromResults(results, sha256="a" * 64, log=self.log)
        iocs = extractor.extract()
        # Should not crash, and should find URL from strings
        url_iocs = [i for i in iocs if i.ioc_type == IOCType.URL]
        assert len(url_iocs) >= 1

    def test_source_type_is_string_for_strings(self):
        results = {
            "strings": [
                {"string": "https://evil.com/test", "string_offset": 42},
            ],
            "decompiled_content": [],
        }
        extractor = IOCExtractorFromResults(results, sha256="a" * 64, log=self.log)
        iocs = extractor.extract()
        url_iocs = [i for i in iocs if i.ioc_type == IOCType.URL]
        assert url_iocs[0].source_type == SourceType.STRING
        assert url_iocs[0].source_identifier == "42"

    def test_source_type_is_decompiled_for_methods(self):
        results = {
            "strings": [],
            "decompiled_content": [
                {
                    "decompiled_method": "https://evil.com/method",
                    "decompiled_method_hash": "myhash",
                    "method_type": "USER",
                }
            ],
        }
        extractor = IOCExtractorFromResults(results, sha256="a" * 64, log=self.log)
        iocs = extractor.extract()
        url_iocs = [i for i in iocs if i.ioc_type == IOCType.URL]
        assert url_iocs[0].source_type == SourceType.DECOMPILED_FUNCTION
        assert url_iocs[0].source_identifier == "myhash"


# ---------------------------------------------------------------------------
# Tests: Worker IOC wiring for APK
# ---------------------------------------------------------------------------

class TestWorkerAPKIOCWiring:
    """Tests that workers.py wires IOC extraction for APK."""

    def test_worker_source_contains_ioc_for_apk(self):
        """Verify workers.py source has IOC extraction in APK branch."""
        import inspect
        from redb import workers
        source = inspect.getsource(workers)
        # APK branch should import IOCExtractorFromResults
        assert "IOCExtractorFromResults" in source
        # Should be in the APK section (near DecompileAPK)
        apk_section = source[source.index("DecompileAPK("):]
        ioc_pos = apk_section.find("IOCExtractorFromResults")
        assert ioc_pos != -1, "IOCExtractorFromResults not found after DecompileAPK"