Xiangjun Kong

"""Unit tests for IOC extraction modules:
- ioc_extractor/ioc_extractor.py — IOCExtractorFromResults
- ioc_extractor/standalone_ioc_extractor.py — IOCScraper
"""
import pytest
from unittest.mock import MagicMock

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


# ============================================================================
# 8a. IOCExtractorFromResults
# ============================================================================

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

    def test_extract_from_strings(self):
        results = {
            "strings": [
                {"string": "Visit https://evil.com/payload", "string_offset": 0x100}
            ],
            "decompiled": [],
        }
        extractor = IOCExtractorFromResults(results, sha256="a" * 64, log=self.log)
        iocs = extractor.extract()
        assert any(ioc.ioc_type == IOCType.URL for ioc in iocs)

    def test_extract_from_decompiled(self):
        results = {
            "strings": [],
            "decompiled": [
                {
                    "decompiled_function": "connect_to('https://c2.malware.org/gate');",
                    "decompiled_function_hash": "hash123",
                    "function_type": "USER",
                }
            ],
        }
        extractor = IOCExtractorFromResults(results, sha256="a" * 64, log=self.log)
        iocs = extractor.extract()
        assert any(ioc.ioc_type == IOCType.URL for ioc in iocs)

    def test_extract_skips_library_functions(self):
        results = {
            "strings": [],
            "decompiled": [
                {
                    "decompiled_function": "call https://should-skip.com",
                    "decompiled_function_hash": "lib_hash",
                    "function_type": "LIBRARY",
                },
                {
                    "decompiled_function": "jmp https://also-skip.com",
                    "decompiled_function_hash": "thunk_hash",
                    "function_type": "THUNK",
                },
            ],
        }
        extractor = IOCExtractorFromResults(results, sha256="a" * 64, log=self.log)
        iocs = extractor.extract()
        assert len(iocs) == 0

    def test_extract_handles_bytes_strings(self):
        results = {
            "strings": [
                {"string": b"Visit https://evil.net/gate", "string_offset": 0}
            ],
            "decompiled": [],
        }
        extractor = IOCExtractorFromResults(results, sha256="a" * 64, log=self.log)
        iocs = extractor.extract()
        assert any(ioc.ioc_type == IOCType.URL for ioc in iocs)

    def test_extract_empty_results(self):
        results = {"strings": [], "decompiled": []}
        extractor = IOCExtractorFromResults(results, sha256="a" * 64, log=self.log)
        iocs = extractor.extract()
        assert len(iocs) == 0

    def test_extract_from_text_raw(self):
        """text_raw entries are scraped and tagged with SourceType.TEXT_RAW."""
        from redb.extractors.ioc_extractor.standalone_ioc_extractor import SourceType

        results = {
            "text_raw": [
                {
                    "content": "var x = fetch('https://attacker.example.org/c2');",
                    "content_hash": "rawhash",
                }
            ],
        }
        extractor = IOCExtractorFromResults(results, sha256="a" * 64, log=self.log)
        iocs = extractor.extract()
        assert any(
            ioc.ioc_type == IOCType.URL and ioc.source_type == SourceType.TEXT_RAW
            for ioc in iocs
        )

    def test_extract_from_text_normalized(self):
        """text_normalized entries are scraped and tagged with TEXT_NORMALIZED.
        This is the surface that catches IOCs hidden behind eval(atob(...)) or
        similar wrappers — the deobfuscator unwraps them, this path scrapes
        the unwrapped form."""
        from redb.extractors.ioc_extractor.standalone_ioc_extractor import SourceType

        results = {
            "text_normalized": [
                {
                    "content": "fetch('http://hidden.example.com/payload');",
                    "content_hash": "normhash",
                }
            ],
        }
        extractor = IOCExtractorFromResults(results, sha256="a" * 64, log=self.log)
        iocs = extractor.extract()
        assert any(
            ioc.ioc_type == IOCType.URL
            and ioc.source_type == SourceType.TEXT_NORMALIZED
            for ioc in iocs
        )

    def test_extract_text_surfaces_distinct_from_decompiled(self):
        """Same URL appearing in raw and normalized forms produces two IOCs
        with distinct source_type values, so analysts can tell them apart."""
        from redb.extractors.ioc_extractor.standalone_ioc_extractor import SourceType

        results = {
            "text_raw": [
                {"content": "https://shared.example.com/", "content_hash": "h1"}
            ],
            "text_normalized": [
                {"content": "https://shared.example.com/", "content_hash": "h2"}
            ],
        }
        extractor = IOCExtractorFromResults(results, sha256="a" * 64, log=self.log)
        iocs = extractor.extract()
        kinds = {ioc.source_type for ioc in iocs if ioc.ioc_type == IOCType.URL}
        assert SourceType.TEXT_RAW in kinds
        assert SourceType.TEXT_NORMALIZED in kinds

    def test_prepare_export_clickhouse(self):
        results = {
            "strings": [
                {"string": "https://evil.com/test", "string_offset": 100}
            ],
            "decompiled": [],
        }
        extractor = IOCExtractorFromResults(results, sha256="a" * 64, log=self.log)
        extractor.extract()
        export = extractor.prepare_export_data("ClickHouseExporter")
        assert export is not None
        data, col_names, col_types = export
        assert len(data) > 0
        assert len(col_names) == 6
        assert len(col_types) == 6

    def test_prepare_export_print(self):
        results = {
            "strings": [
                {"string": "https://evil.com/test", "string_offset": 100}
            ],
            "decompiled": [],
        }
        extractor = IOCExtractorFromResults(results, sha256="a" * 64, log=self.log)
        extractor.extract()
        export = extractor.prepare_export_data("PrintExporter")
        assert isinstance(export, list)
        assert len(export) > 0
        assert "sha256" in export[0]
        assert "ioc_type" in export[0]

    def test_prepare_export_no_iocs(self):
        results = {"strings": [], "decompiled": []}
        extractor = IOCExtractorFromResults(results, sha256="a" * 64, log=self.log)
        extractor.extract()
        export = extractor.prepare_export_data("ClickHouseExporter")
        assert export is None

    def test_get_clickhouse_table(self):
        results = {"strings": [], "decompiled": []}
        extractor = IOCExtractorFromResults(results, sha256="a" * 64, log=self.log)
        assert extractor.get_clickhouse_table() == "redb_iocs"

    def test_tag(self):
        results = {"strings": [], "decompiled": []}
        extractor = IOCExtractorFromResults(results, sha256="a" * 64, log=self.log)
        tag = extractor.tag()
        assert isinstance(tag, str)


# ============================================================================
# 8b. IOCScraper (standalone_ioc_extractor.py)
# ============================================================================

class TestIOCScraperIPv4:
    def setup_method(self):
        self.scraper = IOCScraper()

    def test_scrape_ipv4(self):
        # Use a routable public IP (not in RFC 5737 test ranges, not private)
        text = "Connect to 185.100.87.202 for command"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        ipv4_iocs = [i for i in iocs if i.ioc_type == IOCType.IPV4]
        assert len(ipv4_iocs) == 1
        assert ipv4_iocs[0].ioc_value == "185.100.87.202"

    def test_scrape_ipv4_private_excluded(self):
        text = "192.168.1.1 and 8.8.8.8"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        ipv4_iocs = [i for i in iocs if i.ioc_type == IOCType.IPV4]
        assert len(ipv4_iocs) == 0


class TestIOCScraperIPv6:
    def setup_method(self):
        self.scraper = IOCScraper()

    def test_scrape_ipv6(self):
        text = "Connect to 2001:0db8:85a3:0000:0000:8a2e:0370:7334"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        ipv6_iocs = [i for i in iocs if i.ioc_type == IOCType.IPV6]
        # 2001:0db8 is documentation range, but the regex may still match
        # we just check no crash and proper handling
        assert isinstance(ipv6_iocs, list)


class TestIOCScraperURL:
    def setup_method(self):
        self.scraper = IOCScraper()

    def test_scrape_url(self):
        text = 'load("https://evil.com/payload")'
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        url_iocs = [i for i in iocs if i.ioc_type == IOCType.URL]
        assert len(url_iocs) >= 1
        assert "evil.com" in url_iocs[0].ioc_value

    def test_scrape_defanged_url(self):
        text = "hxxps://evil[.]com/payload"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        url_iocs = [i for i in iocs if i.ioc_type == IOCType.URL]
        assert len(url_iocs) >= 1


class TestIOCScraperEmail:
    def setup_method(self):
        self.scraper = IOCScraper()

    def test_scrape_email(self):
        text = "Send report to [email protected] for review"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        email_iocs = [i for i in iocs if i.ioc_type == IOCType.EMAIL]
        assert len(email_iocs) >= 1
        assert email_iocs[0].ioc_value == "[email protected]"


class TestIOCScraperFQDN:
    def setup_method(self):
        self.scraper = IOCScraper()

    def test_scrape_fqdn(self):
        text = "Resolved evil.com in DNS lookup"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        fqdn_iocs = [i for i in iocs if i.ioc_type == IOCType.FQDN]
        assert any(ioc.ioc_value == "evil.com" for ioc in fqdn_iocs)

    def test_scrape_fqdn_excluded(self):
        text = "Visit example.com"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        fqdn_iocs = [i for i in iocs if i.ioc_type == IOCType.FQDN]
        assert not any(ioc.ioc_value == "example.com" for ioc in fqdn_iocs)


class TestIOCScraperJSContext:
    """Verifies the JS-context FQDN filter rejects JS object-access syntax
    that shape-matches a hostname while preserving real C2 hostnames.

    The corpus below comes from a real Vjw0rm-family screenshot showing 12
    'Domains' extracted from a JS sample of which only 2 were genuine. The
    filter must drop the 10 FPs and keep the 2 TPs.
    """

    def _scrape_fqdns(self, scraper, fragment: str):
        # Wrap each fragment in spaces so the FQDN regex's lookbehind succeeds
        # (the pre-token character is a non-identifier).
        text = f" {fragment} "
        return [
            ioc.ioc_value
            for ioc in scraper.scrape(text, SourceType.STRING, "test")
            if ioc.ioc_type == IOCType.FQDN
        ]

    def test_default_mode_keeps_legacy_behavior(self):
        """Without js_context=True the filter is dormant — every FQDN that
        the regex catches must still pass _validate_fqdn the same way it did
        before this change."""
        scraper = IOCScraper()  # js_context defaults to False
        assert self._scrape_fqdns(scraper, "process.id") == ["process.id"]
        assert self._scrape_fqdns(scraper, "lib.so") == ["lib.so"]
        assert self._scrape_fqdns(scraper, "system.net") == ["system.net"]

    def test_js_context_drops_fp_tlds(self):
        """gTLDs that double as common JS property suffixes (`.name`, `.id`,
        `.so`, `.post`, `.services`, `.tools`, ...) are rejected wholesale
        when js_context is set."""
        scraper = IOCScraper(js_context=True)
        # Each of these is in the screenshot and must be rejected purely on
        # the TLD blocklist (some also hit the SLD blocklist, but the TLD
        # check fires first).
        assert self._scrape_fqdns(scraper, "component.name") == []
        assert self._scrape_fqdns(scraper, "exploit.name") == []
        assert self._scrape_fqdns(scraper, "func.name") == []
        assert self._scrape_fqdns(scraper, "proc.name") == []
        assert self._scrape_fqdns(scraper, "process.name") == []
        assert self._scrape_fqdns(scraper, "lib.so") == []
        assert self._scrape_fqdns(scraper, "proc.id") == []
        assert self._scrape_fqdns(scraper, "http.post") == []
        assert self._scrape_fqdns(scraper, "this.sandboxindicators.services") == []

    def test_js_context_drops_fp_slds(self):
        """JS keywords / framework roots used as the leftmost segment of a
        dotted chain are rejected even when the TLD is legitimate (`system.net`
        is the canonical example: `.net` is a real TLD but `system` is never
        a hostname)."""
        scraper = IOCScraper(js_context=True)
        assert self._scrape_fqdns(scraper, "system.net") == []
        assert self._scrape_fqdns(scraper, "this.foo.com") == []
        assert self._scrape_fqdns(scraper, "process.config.json") == []
        assert self._scrape_fqdns(scraper, "vue.app") == []
        assert self._scrape_fqdns(scraper, "firebase.io") == []

    def test_js_context_preserves_real_c2(self):
        """The screenshot's two true positives must survive: protocol prefix
        SLDs (`ftp`, `smtp`) are intentionally NOT in JS_FP_SLDS so legitimate
        C2 / exfil hostnames keep flowing into the IOC table."""
        scraper = IOCScraper(js_context=True)
        assert self._scrape_fqdns(scraper, "ftp.syfrusvoid.com") == ["ftp.syfrusvoid.com"]
        assert self._scrape_fqdns(scraper, "smtp.gmail.com") == ["smtp.gmail.com"]
        # And a generic malware-style hostname (no TLD/SLD overlap with code)
        # still passes:
        assert self._scrape_fqdns(scraper, "evil-c2.example.org") == ["evil-c2.example.org"]


class TestIOCScraperOnion:
    def setup_method(self):
        self.scraper = IOCScraper()

    def test_scrape_onion(self):
        # v2 onion address (16 chars)
        text = "Connect to expyuzz4wqqyqhjn.onion"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        onion_iocs = [i for i in iocs if i.ioc_type == IOCType.ONION]
        assert len(onion_iocs) >= 1


class TestIOCScraperHashes:
    def setup_method(self):
        self.scraper = IOCScraper()

    def test_scrape_md5_hash(self):
        md5_val = "d41d8cd98f00b204e9800998ecf8427e"
        text = f"Hash: {md5_val}"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        md5_iocs = [i for i in iocs if i.ioc_type == IOCType.HASH_MD5]
        assert len(md5_iocs) == 1

    def test_scrape_sha1_hash(self):
        sha1_val = "da39a3ee5e6b4b0d3255bfef95601890afd80709"
        text = f"Hash: {sha1_val}"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        sha1_iocs = [i for i in iocs if i.ioc_type == IOCType.HASH_SHA1]
        assert len(sha1_iocs) == 1

    def test_scrape_sha256_hash(self):
        sha256_val = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
        text = f"Hash: {sha256_val}"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        sha256_iocs = [i for i in iocs if i.ioc_type == IOCType.HASH_SHA256]
        assert len(sha256_iocs) == 1

    def test_scrape_hash_dedup(self):
        # A SHA256 match should prevent the same hex being emitted as SHA1 substring
        sha256_val = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
        text = f"Hash: {sha256_val}"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        sha256_iocs = [i for i in iocs if i.ioc_type == IOCType.HASH_SHA256]
        sha1_iocs = [i for i in iocs if i.ioc_type == IOCType.HASH_SHA1]
        # SHA256 first 40 chars should not appear as separate SHA1
        assert len(sha256_iocs) >= 1
        first40 = sha256_val[:40]
        assert not any(ioc.ioc_value == first40 for ioc in sha1_iocs)

    def test_scrape_invalid_hash(self):
        # All zeros should be excluded
        text = "Hash: " + "0" * 32
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        hash_iocs = [i for i in iocs if i.ioc_type in (IOCType.HASH_MD5, IOCType.HASH_SHA1, IOCType.HASH_SHA256)]
        assert len(hash_iocs) == 0


class TestIOCScraperCVECWE:
    def setup_method(self):
        self.scraper = IOCScraper()

    def test_scrape_cve(self):
        text = "Exploiting CVE-2021-44228"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        cve_iocs = [i for i in iocs if i.ioc_type == IOCType.CVE]
        assert len(cve_iocs) == 1
        assert cve_iocs[0].ioc_value == "CVE-2021-44228"

    def test_scrape_cwe(self):
        text = "This is CWE-79 in action"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        cwe_iocs = [i for i in iocs if i.ioc_type == IOCType.CWE]
        assert len(cwe_iocs) == 1
        assert cwe_iocs[0].ioc_value == "CWE-79"


class TestIOCScraperCrypto:
    def setup_method(self):
        self.scraper = IOCScraper()

    def test_scrape_btc_address(self):
        # A valid-looking BTC P2PKH address
        text = "Send BTC to 1A1zP1eP5QGefi2DMPTfTL5SLmv7DivfNa"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        btc_iocs = [i for i in iocs if i.ioc_type == IOCType.CRYPTO_BTC]
        assert len(btc_iocs) >= 1

    def test_scrape_eth_address(self):
        text = "ETH wallet: 0xde0B295669a9FD93d5F28D9Ec85E40f4cb697BAe"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        eth_iocs = [i for i in iocs if i.ioc_type == IOCType.CRYPTO_ETH]
        assert len(eth_iocs) >= 1


class TestIOCScraperPaths:
    def setup_method(self):
        self.scraper = IOCScraper()

    def test_scrape_linux_path(self):
        text = "Read config from /etc/passwd"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        path_iocs = [i for i in iocs if i.ioc_type == IOCType.PATH_LINUX]
        assert len(path_iocs) >= 1
        assert any("/etc/passwd" in ioc.ioc_value for ioc in path_iocs)

    def test_scrape_windows_path(self):
        text = r"Load from C:\Windows\System32\cmd.exe"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        path_iocs = [i for i in iocs if i.ioc_type == IOCType.PATH_WINDOWS]
        assert len(path_iocs) >= 1
        assert any(r"C:\Windows\System32\cmd.exe" in ioc.ioc_value for ioc in path_iocs)

    def test_scrape_windows_path_source_escaped(self):
        """Paths embedded in JS / JSON / PowerShell string literals appear with
        doubled backslashes. The regex must accept both forms, and the stored
        IOC must be normalised so escaped and runtime forms collapse to one."""
        text = "const exclusions = ['C:\\\\Windows\\\\Temp', 'C:\\\\Users\\\\Public'];"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        path_iocs = [i for i in iocs if i.ioc_type == IOCType.PATH_WINDOWS]
        assert len(path_iocs) >= 2
        values = {ioc.ioc_value for ioc in path_iocs}
        assert r"C:\Windows\Temp" in values
        assert r"C:\Users\Public" in values
        assert not any("\\\\" in v for v in values), f"backslashes not normalised: {values}"

    def test_scrape_windows_path_with_wildcard(self):
        """Malware commonly uses wildcard paths like the Defender-exclusion
        pattern below; `*` is a valid path component and must be preserved."""
        text = "'C:\\\\Users\\\\*\\\\AppData\\\\Local\\\\Temp'"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        path_iocs = [i for i in iocs if i.ioc_type == IOCType.PATH_WINDOWS]
        assert any(r"C:\Users\*\AppData\Local\Temp" == ioc.ioc_value for ioc in path_iocs)

    # --- Behaviour-pinning regression tests -----------------------------------
    # These lock in the trade-offs of the two-tier component grammar (strict
    # first char, permissive body in delimited segments, strict final segment,
    # colon excluded from bodies). Every one of them was a real-or-potential
    # regression while the regex was being rewritten.

    def test_path_with_internal_space(self):
        """`Program Files` is the canonical reason segment bodies must allow
        whitespace. The old regex's permissive body class was correct on this
        case; the new grammar must preserve it."""
        text = r"Path: C:\Program Files\Microsoft\app.exe"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        path_iocs = [i for i in iocs if i.ioc_type == IOCType.PATH_WINDOWS]
        assert any(r"C:\Program Files\Microsoft\app.exe" == ioc.ioc_value for ioc in path_iocs)

    def test_path_with_internal_space_in_escaped_form(self):
        """The combination of source-escaping and internal whitespace is the
        case the old regex got wrong (it required single backslashes) and the
        first iteration of the new regex got wrong (it forbade whitespace
        everywhere). Both must work now."""
        text = "'C:\\\\Program Files (x86)\\\\Microsoft\\\\app.exe'"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        path_iocs = [i for i in iocs if i.ioc_type == IOCType.PATH_WINDOWS]
        assert any(r"C:\Program Files (x86)\Microsoft\app.exe" == ioc.ioc_value for ioc in path_iocs)

    def test_path_stops_at_whitespace_when_no_separator_follows(self):
        """The strict final-segment class prevents prose-slurping. Critical
        because segment bodies allow whitespace — without this, every path
        followed by free text would gobble the rest of the line."""
        text = r"see c:\users\admin and other stuff"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        path_iocs = [i for i in iocs if i.ioc_type == IOCType.PATH_WINDOWS]
        assert any(ioc.ioc_value == r"c:\users\admin" for ioc in path_iocs)
        assert not any("stuff" in ioc.ioc_value for ioc in path_iocs)

    def test_two_drives_separated_by_prose_match_independently(self):
        """`From C:\\one to D:\\two` must yield two IOCs, not one slurp.
        Pinned because excluding `:` from segment bodies is what enables this
        — without that exclusion, ` to D:` would be valid body of segment 1."""
        text = r"From C:\one to D:\two"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        values = {i.ioc_value for i in iocs if i.ioc_type == IOCType.PATH_WINDOWS}
        assert r"C:\one" in values
        assert r"D:\two" in values

    def test_path_lowercase_drive(self):
        text = r"Backup at d:\backup\db.sql"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        path_iocs = [i for i in iocs if i.ioc_type == IOCType.PATH_WINDOWS]
        assert any(ioc.ioc_value == r"d:\backup\db.sql" for ioc in path_iocs)

    def test_path_just_root(self):
        """Bare `C:\\` (the root of a drive) must match — Defender exclusion
        lists ship it as a literal entry."""
        text = "['C:\\\\', 'C:\\\\Windows']"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        values = {i.ioc_value for i in iocs if i.ioc_type == IOCType.PATH_WINDOWS}
        assert "C:\\" in values
        assert r"C:\Windows" in values

    def test_drive_letter_alone_does_not_match(self):
        """`D:` with no separator following is not a path."""
        text = "Drive D: is mounted"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        path_iocs = [i for i in iocs if i.ioc_type == IOCType.PATH_WINDOWS]
        assert path_iocs == []

    def test_trailing_punctuation_is_stripped(self):
        """`See C:\\Path\\file.txt.` — the final period is sentence
        punctuation, not part of the path. `rstrip('.,;:')` strips it."""
        text = r"See C:\Path\file.txt, and also C:\Other\file.exe."
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        values = {i.ioc_value for i in iocs if i.ioc_type == IOCType.PATH_WINDOWS}
        assert r"C:\Path\file.txt" in values
        assert r"C:\Other\file.exe" in values


class TestIOCScraperRegistryKey:
    def setup_method(self):
        self.scraper = IOCScraper()

    def test_scrape_hklm_short_form(self):
        text = r"reg add HKLM\SYSTEM\CurrentControlSet\Services\SharedAccess /v EnableFirewall"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        reg_iocs = [i for i in iocs if i.ioc_type == IOCType.REGISTRY_KEY]
        assert any(r"HKLM\SYSTEM\CurrentControlSet\Services\SharedAccess" in ioc.ioc_value for ioc in reg_iocs)

    def test_scrape_hkey_long_form(self):
        text = r"HKEY_CURRENT_USER\Software\Microsoft\Windows\CurrentVersion\Run"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        reg_iocs = [i for i in iocs if i.ioc_type == IOCType.REGISTRY_KEY]
        assert any(r"HKEY_CURRENT_USER\Software\Microsoft\Windows\CurrentVersion\Run" == ioc.ioc_value for ioc in reg_iocs)

    def test_scrape_registry_source_escaped(self):
        """Same source-escaping concern as Windows paths."""
        text = "'HKLM\\\\SYSTEM\\\\CurrentControlSet\\\\Services\\\\SharedAccess\\\\Parameters\\\\FirewallPolicy\\\\DomainProfile'"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        reg_iocs = [i for i in iocs if i.ioc_type == IOCType.REGISTRY_KEY]
        assert any(r"HKLM\SYSTEM\CurrentControlSet\Services\SharedAccess\Parameters\FirewallPolicy\DomainProfile" == ioc.ioc_value for ioc in reg_iocs)
        assert all("\\\\" not in ioc.ioc_value for ioc in reg_iocs)

    def test_bare_hive_not_extracted(self):
        """A bare hive mention with no path component should not match — too
        common in prose ('the HKLM hive') to be useful as an IOC."""
        text = "The HKLM hive contains system-wide settings; HKCU is per-user."
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        reg_iocs = [i for i in iocs if i.ioc_type == IOCType.REGISTRY_KEY]
        assert reg_iocs == []

    def test_registry_not_classified_as_path(self):
        """HKLM\\... starts with letters not a drive-letter+colon, so the
        Windows path regex must not also match it."""
        text = r"HKLM\SYSTEM\CurrentControlSet\Services\Foo"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        path_iocs = [i for i in iocs if i.ioc_type == IOCType.PATH_WINDOWS]
        assert path_iocs == []


class TestIOCScraperServer:
    def setup_method(self):
        self.scraper = IOCScraper()

    def test_scrape_server(self):
        text = "Connect to 185.100.87.202:8080 for C2"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        server_iocs = [i for i in iocs if i.ioc_type == IOCType.SERVER]
        assert len(server_iocs) >= 1
        assert "185.100.87.202:8080" in server_iocs[0].ioc_value


class TestIOCScraperDedup:
    def setup_method(self):
        self.scraper = IOCScraper()

    def test_scrape_dedup_within_text(self):
        text = "https://evil.com https://evil.com"
        iocs = list(self.scraper.scrape(text, SourceType.STRING, "test"))
        url_iocs = [i for i in iocs if i.ioc_type == IOCType.URL]
        assert len(url_iocs) == 1

    def test_scrape_empty_text(self):
        iocs = list(self.scraper.scrape("", SourceType.STRING, "test"))
        assert len(iocs) == 0