Deduplicate AndroidQF SMS analysis

docs/iocs.md

@@ -37,10 +37,8 @@ export MVT_STIX2="/home/user/IOC1.stix2:/home/user/IOC2.stix2"
 ## Network Access
 
 When checking URL indicators, MVT follows recognized shortened URLs with an
-HTTP `HEAD` request. URL checks are deduplicated and run concurrently, with at
-most 20 requests in progress at a time. Redirects within an individual URL
-chain are still followed sequentially. The following environment variables
-control these requests:
+HTTP `HEAD` request. The following environment variables control these
+requests:
 
 - `MVT_NETWORK_ACCESS_ALLOWED` enables or disables network requests. It defaults
   to `true`. Set it to `false` to prevent MVT from attempting to resolve
@@ -75,3 +73,4 @@ You can automaticallly download the latest public indicator files with the comma
 
 Please [open an issue](https://github.com/mvt-project/mvt/issues/) to suggest new sources of STIX-formatted IOCs.
 
+

src/mvt/android/artifacts/dumpsys_adb.py

@@ -13,13 +13,6 @@ from .artifact import AndroidArtifact
 class DumpsysADBArtifact(AndroidArtifact):
     multiline_fields = ["user_keys", "keystore"]
 
-    @staticmethod
-    def is_structural_line(key: str, vals: list) -> bool:
-        if key == "}":
-            return True
-        # XML keystore continuations also split on "=", but never into an identifier.
-        return len(vals) == 2 and key.isidentifier()
-
     def indented_dump_parser(self, dump_data):
         """
         Parse the indented dumpsys output, generated by DualDumpOutputStream in Android.
@@ -48,18 +41,10 @@ class DumpsysADBArtifact(AndroidArtifact):
                 if key == "":
                     # If the line is empty, it's the terminator for the multiline value
                     in_multiline = False
-                    if isinstance(stack[-1], list):
-                        stack.pop()
-                    continue
-
-                if not self.is_structural_line(key, vals):
-                    current_dict.append(line.lstrip())
-                    continue
-
-                in_multiline = False
-                if isinstance(stack[-1], list):
                     stack.pop()
-                current_dict = stack[-1]
+                else:
+                    current_dict.append(line.lstrip())
+                continue
 
             if key == "}":
                 stack.pop()

src/mvt/android/modules/backup/sms.py

@@ -36,8 +36,6 @@ class SMS(BackupModule):
         if not self.indicators:
             return
 
-        messages = []
-        url_batches = []
         for message in self.results:
             if "body" not in message:
                 continue
@@ -46,16 +44,12 @@ class SMS(BackupModule):
             if message_links == []:
                 message_links = check_for_links(message.get("text", ""))
 
-            messages.append(message)
-            url_batches.append(message_links)
-
-        for message, ioc_match in zip(
-            messages, self.indicators.check_url_batches(url_batches)
-        ):
+            ioc_match = self.indicators.check_urls(message_links)
             if ioc_match:
                 self.alertstore.critical(
                     ioc_match.message, "", message, matched_indicator=ioc_match.ioc
                 )
+                continue
 
     def run(self) -> None:
         sms_path = "apps/com.android.providers.telephony/d_f/*_sms_backup"

src/mvt/common/indicators.py

@@ -7,10 +7,9 @@ import glob
 import json
 import logging
 import os
-from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass
 from functools import lru_cache
-from typing import Any, Dict, Iterator, List, Optional, Sequence
+from typing import Any, Dict, Iterator, List, Optional
 
 import ahocorasick
 from appdirs import user_data_dir
@@ -23,8 +22,6 @@ MVT_INDICATORS_FOLDER = os.path.join(MVT_DATA_FOLDER, "indicators")
 
 logger = logging.getLogger(__name__)
 
-URL_CHECK_MAX_WORKERS = 20
-
 
 @dataclass
 class Indicator:
@@ -493,41 +490,12 @@ class Indicators:
         if not urls:
             return None
 
-        return self.check_url_batches([urls])[0]
+        for url in urls:
+            check = self.check_url(url)
+            if check:
+                return check
 
-    def check_url_batches(
-        self, url_batches: Sequence[Optional[Sequence[str]]]
-    ) -> List[Optional[IndicatorMatch]]:
-        """Check batches of URLs concurrently while preserving batch order.
-
-        URLs are deduplicated across batches before checking. Each returned item
-        is the first indicator match from the corresponding input batch, using
-        the original URL order.
-        """
-        batches = [list(urls) if urls else [] for urls in url_batches]
-        unique_urls = list(
-            dict.fromkeys(url for urls in batches for url in urls)
-        )
-
-        if not unique_urls:
-            return [None] * len(batches)
-
-        if settings.NETWORK_ACCESS_ALLOWED and len(unique_urls) > 1:
-            worker_count = min(URL_CHECK_MAX_WORKERS, len(unique_urls))
-            with ThreadPoolExecutor(max_workers=worker_count) as executor:
-                url_matches = dict(
-                    zip(unique_urls, executor.map(self.check_url, unique_urls))
-                )
-        else:
-            url_matches = {url: self.check_url(url) for url in unique_urls}
-
-        return [
-            next(
-                (url_matches[url] for url in urls if url_matches[url] is not None),
-                None,
-            )
-            for urls in batches
-        ]
+        return None
 
     def check_process(self, process: str) -> Optional[IndicatorMatch]:
         """Check the provided process name against the list of process

src/mvt/common/url.py

@@ -5,7 +5,6 @@
 
 import logging
 from typing import Optional
-from urllib.parse import urlparse
 
 import requests
 from tld import get_tld
@@ -374,10 +373,6 @@ class URL:
         :rtype: bool
 
         """
-        parsed_url = urlparse(self.url if "://" in self.url else f"//{self.url}")
-        if self.domain.lower() == "goo.gl" and parsed_url.path.startswith("/maps/"):
-            return False
-
         if self.domain.lower() in SHORTENER_DOMAINS:
             self.is_shortened = True
 

src/mvt/ios/modules/mixed/shortcuts.py

@@ -76,10 +76,8 @@ class Shortcuts(IOSExtraction):
         if not self.indicators:
             return
 
-        url_batches = [result["action_urls"] for result in self.results]
-        for result, ioc_match in zip(
-            self.results, self.indicators.check_url_batches(url_batches)
-        ):
+        for result in self.results:
+            ioc_match = self.indicators.check_urls(result["action_urls"])
             if ioc_match:
                 self.alertstore.critical(
                     ioc_match.message, "", result, matched_indicator=ioc_match.ioc

src/mvt/ios/modules/mixed/sms.py

@@ -85,17 +85,12 @@ class SMS(IOSExtraction):
         if not self.indicators:
             return
 
-        url_batches = []
         for result in self.results:
             message_links = result.get("links", [])
             # Making sure not link was ignored
             if message_links == []:
                 message_links = check_for_links(result.get("text", ""))
-            url_batches.append(message_links)
-
-        for result, ioc_match in zip(
-            self.results, self.indicators.check_url_batches(url_batches)
-        ):
+            ioc_match = self.indicators.check_urls(message_links)
             if ioc_match:
                 self.alertstore.critical(
                     ioc_match.message, "", result, matched_indicator=ioc_match.ioc

src/mvt/ios/modules/mixed/webkit_session_resource_log.py

@@ -70,8 +70,6 @@ class WebkitSessionResourceLog(IOSExtraction):
         if not self.indicators:
             return
 
-        records = []
-        url_batches = []
         for _, entries in self.results.items():
             for entry in entries:
                 source_domains = self._extract_domains(entry["redirect_source"])
@@ -96,42 +94,36 @@ class WebkitSessionResourceLog(IOSExtraction):
                     )
                 )
 
-                records.append((entry, source_domains, destination_domains))
-                url_batches.append(all_origins)
+                ioc_match = self.indicators.check_urls(all_origins)
+                if ioc_match:
+                    self.alertstore.critical(
+                        ioc_match.message, "", entry, matched_indicator=ioc_match.ioc
+                    )
 
-        for record, ioc_match in zip(
-            records, self.indicators.check_url_batches(url_batches)
-        ):
-            if ioc_match:
-                entry, source_domains, destination_domains = record
-                self.alertstore.critical(
-                    ioc_match.message, "", entry, matched_indicator=ioc_match.ioc
-                )
+                    redirect_path = ""
+                    if len(source_domains) > 0:
+                        redirect_path += "SOURCE: "
+                        for idx, item in enumerate(source_domains):
+                            source_domains[idx] = f'"{item}"'
 
-                redirect_path = ""
-                if len(source_domains) > 0:
-                    redirect_path += "SOURCE: "
-                    for idx, item in enumerate(source_domains):
-                        source_domains[idx] = f'"{item}"'
+                        redirect_path += ", ".join(source_domains)
+                        redirect_path += " -> "
 
-                    redirect_path += ", ".join(source_domains)
-                    redirect_path += " -> "
+                    redirect_path += f'ORIGIN: "{entry["origin"]}"'
 
-                redirect_path += f'ORIGIN: "{entry["origin"]}"'
+                    if len(destination_domains) > 0:
+                        redirect_path += " -> "
+                        redirect_path += "DESTINATION: "
+                        for idx, item in enumerate(destination_domains):
+                            destination_domains[idx] = f'"{item}"'
 
-                if len(destination_domains) > 0:
-                    redirect_path += " -> "
-                    redirect_path += "DESTINATION: "
-                    for idx, item in enumerate(destination_domains):
-                        destination_domains[idx] = f'"{item}"'
+                        redirect_path += ", ".join(destination_domains)
 
-                    redirect_path += ", ".join(destination_domains)
-
-                self.alertstore.high(
-                    f"Found HTTP redirect between suspicious domains: {redirect_path}",
-                    "",
-                    entry,
-                )
+                    self.alertstore.high(
+                        f"Found HTTP redirect between suspicious domains: {redirect_path}",
+                        "",
+                        entry,
+                    )
 
     def _extract_browsing_stats(self, log_path):
         items = []

src/mvt/ios/modules/mixed/whatsapp.py

@@ -61,10 +61,8 @@ class Whatsapp(IOSExtraction):
         if not self.indicators:
             return
 
-        url_batches = [result.get("links", []) for result in self.results]
-        for result, ioc_match in zip(
-            self.results, self.indicators.check_url_batches(url_batches)
-        ):
+        for result in self.results:
+            ioc_match = self.indicators.check_urls(result.get("links", []))
             if ioc_match:
                 self.alertstore.critical(
                     ioc_match.message, "", result, matched_indicator=ioc_match.ioc

tests/android/test_artifact_dumpsys_adb.py

@@ -30,66 +30,6 @@ class TestDumpsysADBArtifact:
         )
         assert user_key["user"] == "user@linux"
 
-    def test_parsing_adb_wifi(self):
-        da_adb = DumpsysADBArtifact()
-        file = get_artifact("android_data/dumpsys_adb_wifi.txt")
-        with open(file, "rb") as f:
-            data = f.read()
-
-        da_adb.parse(data)
-
-        assert len(da_adb.results) == 1
-        adb_data = da_adb.results[0]
-        assert "user_keys" in adb_data
-        assert len(adb_data["user_keys"]) == 1
-
-        user_key = adb_data["user_keys"][0]
-        assert (
-            user_key["fingerprint"] == "F0:A1:3D:8C:B3:F4:7B:09:9F:EE:8B:D8:38:2E:BD:C6"
-        )
-        assert user_key["user"] == "user@linux"
-
-        # The adb_wifi block following the keystore is not part of the keystore.
-        assert b"adb_wifi" not in adb_data["keystore"]
-
-    def test_parsing_multiline_terminated_by_structural_line(self):
-        dump_data = (
-            b"debugging_manager={\n"
-            b"  keystore=ABX\x00\x0bkeyStore\x00\x02\x11\n"
-            b"  connected_to_adb=true\n"
-            b"  adb_wifi={\n"
-            b"    enabled=false\n"
-            b"    tls_port=0\n"
-            b"  }\n"
-        )
-
-        parsed = DumpsysADBArtifact().indented_dump_parser(dump_data)
-
-        debugging_manager = parsed["debugging_manager"]
-        assert debugging_manager["keystore"] == [b"ABX\x00\x0bkeyStore\x00\x02\x11"]
-        assert debugging_manager["connected_to_adb"] == b"true"
-        assert debugging_manager["adb_wifi"] == {
-            "enabled": b"false",
-            "tls_port": b"0",
-        }
-
-    def test_parsing_multiline_terminated_by_closing_brace(self):
-        dump_data = (
-            b"debugging_manager={\n"
-            b"  keystore=ABX\x00\x0bkeyStore\x00\x02\x11\n"
-            b"}\n"
-            b"other={\n"
-            b"  value=true\n"
-            b"}\n"
-        )
-
-        parsed = DumpsysADBArtifact().indented_dump_parser(dump_data)
-
-        assert parsed["debugging_manager"]["keystore"] == [
-            b"ABX\x00\x0bkeyStore\x00\x02\x11"
-        ]
-        assert parsed["other"] == {"value": b"true"}
-
     def test_parsing_adb_xml(self):
         da_adb = DumpsysADBArtifact()
         file = get_artifact("android_data/dumpsys_adb_xml.txt")

tests/common/test_indicators.py

@@ -5,9 +5,7 @@
 
 import logging
 import os
-import threading
 
-import requests
 
 from mvt.common.config import settings
 from mvt.common.indicators import Indicators
@@ -82,127 +80,6 @@ class TestIndicators:
         assert ind.check_url("https://198.51.100.1:8080/")
         assert ind.check_url("https://1.1.1.1/") is None
 
-    def test_google_maps_short_url_is_not_resolved(self, indicator_file, mocker):
-        head_request = mocker.patch("mvt.common.url.requests.head")
-        ind = Indicators(log=logging)
-        ind.load_indicators_files([indicator_file], load_default=False)
-
-        assert ind.check_url("https://goo.gl/maps/example") is None
-        head_request.assert_not_called()
-
-    def test_check_url_batches_preserves_order(self, indicator_file):
-        ind = Indicators(log=logging)
-        ind.load_indicators_files([indicator_file], load_default=False)
-
-        matches = ind.check_url_batches(
-            [
-                [
-                    "https://github.com",
-                    "http://example.com/thisisbad",
-                    "https://www.example.org/foobar",
-                ],
-                ["https://github.com", "https://www.example.org/foobar"],
-                [],
-                None,
-            ]
-        )
-
-        assert matches[0]
-        assert matches[0].ioc.value == "http://example.com/thisisbad"
-        assert matches[1]
-        assert matches[1].ioc.value == "example.org"
-        assert matches[2] is None
-        assert matches[3] is None
-
-    def test_check_url_batches_deduplicates_and_limits_workers(
-        self, indicator_file, mocker
-    ):
-        ind = Indicators(log=logging)
-        ind.load_indicators_files([indicator_file], load_default=False)
-        mocker.patch("mvt.common.indicators.URL_CHECK_MAX_WORKERS", 2)
-
-        barrier = threading.Barrier(2)
-        lock = threading.Lock()
-        calls = []
-        active = 0
-        max_active = 0
-
-        def head_request(url, timeout):
-            nonlocal active, max_active
-            with lock:
-                calls.append(url)
-                active += 1
-                max_active = max(max_active, active)
-                call_number = len(calls)
-
-            try:
-                if call_number <= 2:
-                    barrier.wait(timeout=5)
-                return mocker.Mock(status_code=200, headers={})
-            finally:
-                with lock:
-                    active -= 1
-
-        mocker.patch("mvt.common.url.requests.head", side_effect=head_request)
-        urls = [
-            "https://bit.ly/one",
-            "https://tinyurl.com/two",
-            "https://t.co/three",
-        ]
-
-        assert ind.check_url_batches([urls, [urls[0]]]) == [None, None]
-        assert sorted(calls) == sorted(urls)
-        assert max_active == 2
-
-    def test_check_url_batches_respects_disabled_network(
-        self, indicator_file, mocker
-    ):
-        ind = Indicators(log=logging)
-        ind.load_indicators_files([indicator_file], load_default=False)
-        mocker.patch("mvt.common.indicators.settings.NETWORK_ACCESS_ALLOWED", False)
-        head_request = mocker.patch("mvt.common.url.requests.head")
-
-        assert ind.check_url_batches([["https://bit.ly/example"]]) == [None]
-        head_request.assert_not_called()
-
-    def test_check_url_batches_handles_nested_redirects_and_request_failures(
-        self, indicator_file, mocker
-    ):
-        ind = Indicators(log=logging)
-        ind.load_indicators_files([indicator_file], load_default=False)
-
-        def head_request(url, timeout):
-            if url == "https://bit.ly/failure":
-                raise requests.Timeout()
-            if url == "https://tinyurl.com/nested":
-                return mocker.Mock(
-                    status_code=301,
-                    headers={"Location": "https://t.co/nested"},
-                )
-            if url == "https://t.co/nested":
-                return mocker.Mock(
-                    status_code=302,
-                    headers={"Location": "https://www.example.org/landing"},
-                )
-            raise AssertionError(f"Unexpected URL: {url}")
-
-        head = mocker.patch(
-            "mvt.common.url.requests.head", side_effect=head_request
-        )
-
-        matches = ind.check_url_batches(
-            [["https://bit.ly/failure"], ["https://tinyurl.com/nested"]]
-        )
-
-        assert matches[0] is None
-        assert matches[1]
-        assert matches[1].ioc.value == "example.org"
-        assert {call.args[0] for call in head.call_args_list} == {
-            "https://bit.ly/failure",
-            "https://tinyurl.com/nested",
-            "https://t.co/nested",
-        }
-
     def test_check_file_hash(self, indicator_file):
         ind = Indicators(log=logging)
         ind.load_indicators_files([indicator_file], load_default=False)

tests/common/test_url.py

@@ -1,24 +0,0 @@
-# Mobile Verification Toolkit (MVT)
-# Copyright (c) 2021-2023 The MVT Authors.
-# Use of this software is governed by the MVT License 1.1 that can be found at
-#   https://license.mvt.re/1.1/
-
-import pytest
-
-from mvt.common.url import URL
-
-
-@pytest.mark.parametrize(
-    "url",
-    [
-        "https://goo.gl/maps/example",
-        "http://goo.gl/maps/example?entry=message",
-        "goo.gl/maps/example",
-    ],
-)
-def test_google_maps_url_is_not_shortened(url):
-    assert URL(url).check_if_shortened() is False
-
-
-def test_other_google_short_url_is_shortened():
-    assert URL("https://goo.gl/example").check_if_shortened() is True

tests/ios_backup/test_sms.py

@@ -29,31 +29,3 @@ class TestSMSModule:
         m.indicators = ind
         run_module(m)
         assert len(m.alertstore.alerts) == 1
-
-    def test_detection_batches_urls_and_preserves_event(self, indicator_file, mocker):
-        results = [
-            {
-                "text": "first",
-                "links": ["http://example.com/thisisbad"],
-            },
-            {
-                "text": "second",
-                "links": ["https://github.com"],
-            },
-        ]
-        m = SMS(results=results)
-        ind = Indicators(log=logging.getLogger())
-        ind.parse_stix2(indicator_file)
-        batch_check = mocker.spy(ind, "check_url_batches")
-        m.indicators = ind
-
-        m.check_indicators()
-
-        batch_check.assert_called_once_with(
-            [
-                ["http://example.com/thisisbad"],
-                ["https://github.com"],
-            ]
-        )
-        assert len(m.alertstore.alerts) == 1
-        assert m.alertstore.alerts[0].event is results[0]