Adding indicator to matched results

This way when a detection is logged, the user can know which STIX2
file was matched by the module

.github/workflows/python-package.yml

@@ -16,7 +16,8 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: [3.7, 3.8, 3.9]
+        # python-version: [3.7, 3.8, 3.9]
+        python-version: [3.8, 3.9]
 
     steps:
     - uses: actions/checkout@v2
@@ -27,8 +28,9 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        python -m pip install flake8 pytest safety
+        python -m pip install flake8 pytest safety stix2
         if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+        python -m pip install .
     - name: Lint with flake8
       run: |
         # stop the build if there are Python syntax errors or undefined names
@@ -37,7 +39,5 @@ jobs:
         flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
     - name: Safety checks
       run: safety check
-
+    - name: Test with pytest
-    # - name: Test with pytest
+      run: pytest
-    #   run: |
-    #     pytest

docs/iocs.md

@@ -38,7 +38,9 @@ export MVT_STIX2="/home/user/IOC1.stix2:/home/user/IOC2.stix2"
 
 - The [Amnesty International investigations repository](https://github.com/AmnestyTech/investigations) contains STIX-formatted IOCs for:
     - [Pegasus](https://en.wikipedia.org/wiki/Pegasus_(spyware)) ([STIX2](https://raw.githubusercontent.com/AmnestyTech/investigations/master/2021-07-18_nso/pegasus.stix2))
-    - [Predator from Cytrox](https://citizenlab.ca/2021/12/pegasus-vs-predator-dissidents-doubly-infected-iphone-reveals-cytrox-mercenary-spyware/) ([STIX2](https://github.com/AmnestyTech/investigations/tree/master/2021-12-16_cytrox/cytrox.stix2))
+    - [Predator from Cytrox](https://citizenlab.ca/2021/12/pegasus-vs-predator-dissidents-doubly-infected-iphone-reveals-cytrox-mercenary-spyware/) ([STIX2](https://raw.githubusercontent.com/AmnestyTech/investigations/master/2021-12-16_cytrox/cytrox.stix2))
-- [This repository](https://github.com/Te-k/stalkerware-indicators) contains IOCs for Android stalkerware including [a STIX MVT-compatible file](https://github.com/Te-k/stalkerware-indicators/blob/master/stalkerware.stix2).
+- [This repository](https://github.com/Te-k/stalkerware-indicators) contains IOCs for Android stalkerware including [a STIX MVT-compatible file](https://raw.githubusercontent.com/Te-k/stalkerware-indicators/master/stalkerware.stix2).
+
+You can automaticallly download the latest public indicator files with the command `mvt-ios download-iocs` or `mvt-android download-iocs`. These commands download the list of indicators listed [here](https://github.com/mvt-project/mvt/blob/main/public_indicators.json) and store them in the [appdir](https://pypi.org/project/appdirs/) folder. They are then loaded automatically by mvt.
 
 Please [open an issue](https://github.com/mvt-project/mvt/issues/) to suggest new sources of STIX-formatted IOCs.

mvt/android/cli.py

@@ -9,10 +9,10 @@ import os
 import click
 from rich.logging import RichHandler
 
-from mvt.common.help import HELP_MSG_MODULE, HELP_MSG_IOC
+from mvt.common.help import (HELP_MSG_FAST, HELP_MSG_IOC,
-from mvt.common.help import HELP_MSG_FAST, HELP_MSG_OUTPUT, HELP_MSG_LIST_MODULES
+                             HELP_MSG_LIST_MODULES, HELP_MSG_MODULE,
-from mvt.common.help import HELP_MSG_SERIAL
+                             HELP_MSG_OUTPUT, HELP_MSG_SERIAL)
-from mvt.common.indicators import Indicators, IndicatorsFileBadFormat
+from mvt.common.indicators import Indicators, download_indicators_files
 from mvt.common.logo import logo
 from mvt.common.module import run_module, save_timeline
 
@@ -129,13 +129,7 @@ def check_adb(ctx, iocs, output, fast, list_modules, module, serial):
             ctx.exit(1)
 
     indicators = Indicators(log=log)
-    for ioc_path in iocs:
+    indicators.load_indicators_files(iocs)
-        try:
-            indicators.parse_stix2(ioc_path)
-        except IndicatorsFileBadFormat as e:
-            log.critical(e)
-            ctx.exit(1)
-    log.info("Loaded a total of %d indicators", indicators.ioc_count)
 
     timeline = []
     timeline_detected = []
@@ -145,13 +139,12 @@ def check_adb(ctx, iocs, output, fast, list_modules, module, serial):
 
         m = adb_module(output_folder=output, fast_mode=fast,
                        log=logging.getLogger(adb_module.__module__))
+        if indicators.total_ioc_count:
+            m.indicators = indicators
+            m.indicators.log = m.log
         if serial:
             m.serial = serial
 
-        if iocs:
-            indicators.log = m.log
-            m.indicators = indicators
-
         run_module(m)
         timeline.extend(m.timeline)
         timeline_detected.extend(m.timeline_detected)
@@ -184,13 +177,7 @@ def check_backup(ctx, iocs, output, backup_path, serial):
             ctx.exit(1)
 
     indicators = Indicators(log=log)
-    for ioc_path in iocs:
+    indicators.load_indicators_files(iocs)
-        try:
-            indicators.parse_stix2(ioc_path)
-        except IndicatorsFileBadFormat as e:
-            log.critical(e)
-            ctx.exit(1)
-    log.info("Loaded a total of %d indicators", indicators.ioc_count)
 
     if os.path.isfile(backup_path):
         log.critical("The path you specified is a not a folder!")
@@ -203,12 +190,18 @@ def check_backup(ctx, iocs, output, backup_path, serial):
     for module in BACKUP_MODULES:
         m = module(base_folder=backup_path, output_folder=output,
                    log=logging.getLogger(module.__module__))
-
+        if indicators.total_ioc_count:
+            m.indicators = indicators
+            m.indicators.log = m.log
         if serial:
             m.serial = serial
 
-        if len(indicators.ioc_count) > 0:
-            indicators.log = m.log
-            m.indicators = indicators
-
         run_module(m)
+
+
+#==============================================================================
+# Command: download-iocs
+#==============================================================================
+@cli.command("download-iocs", help="Download public STIX2 indicators")
+def download_indicators():
+    download_indicators_files(log)

mvt/android/modules/adb/__init__.py

@@ -4,6 +4,7 @@
 #   https://license.mvt.re/1.1/
 
 from .chrome_history import ChromeHistory
+from .dumpsys_accessibility import DumpsysAccessibility
 from .dumpsys_batterystats import DumpsysBatterystats
 from .dumpsys_full import DumpsysFull
 from .dumpsys_packages import DumpsysPackages
@@ -18,6 +19,6 @@ from .sms import SMS
 from .whatsapp import Whatsapp
 
 ADB_MODULES = [ChromeHistory, SMS, Whatsapp, Processes,
-               DumpsysBatterystats, DumpsysProcstats,
+               DumpsysAccessibility, DumpsysBatterystats, DumpsysProcstats,
                DumpsysPackages, DumpsysReceivers, DumpsysFull,
                Packages, RootBinaries, Logcat, Files]

mvt/android/modules/adb/base.py

@@ -112,7 +112,7 @@ class AndroidExtraction(MVTModule):
         :returns: Output of command
 
         """
-        return self.device.shell(command)
+        return self.device.shell(command, read_timeout_s=200.0)
 
     def _adb_check_if_root(self):
         """Check if we have a `su` binary on the Android device.

mvt/android/modules/adb/dumpsys_accessibility.py

@@ -0,0 +1,53 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021 The MVT Project 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 io
+import logging
+import os
+
+from .base import AndroidExtraction
+
+log = logging.getLogger(__name__)
+
+
+class DumpsysAccessibility(AndroidExtraction):
+    """This module extracts stats on accessibility."""
+
+    def __init__(self, file_path=None, base_folder=None, output_folder=None,
+                 serial=None, fast_mode=False, log=None, results=[]):
+        super().__init__(file_path=file_path, base_folder=base_folder,
+                         output_folder=output_folder, fast_mode=fast_mode,
+                         log=log, results=results)
+
+    def run(self):
+        self._adb_connect()
+
+        stats = self._adb_command("dumpsys accessibility")
+
+        in_services = False
+        for line in stats.split("\n"):
+            if line.strip().startswith("installed services:"):
+                in_services = True
+                continue
+
+            if not in_services:
+                continue
+
+            if line.strip() == "}":
+                break
+
+            service = line.split(":")[1].strip()
+            log.info("Found installed accessibility service \"%s\"", service)
+
+        if self.output_folder:
+            acc_path = os.path.join(self.output_folder,
+                                      "dumpsys_accessibility.txt")
+            with io.open(acc_path, "w", encoding="utf-8") as handle:
+                handle.write(stats)
+
+            log.info("Records from dumpsys accessibility stored at %s",
+                     acc_path)
+
+        self._adb_disconnect()

mvt/android/modules/adb/files.py

@@ -3,12 +3,11 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
-import logging
-import os
-import stat
 import datetime
+import logging
+import stat
 
-from mvt.common.utils import check_for_links, convert_timestamp_to_iso
+from mvt.common.utils import convert_timestamp_to_iso
 
 from .base import AndroidExtraction
 
@@ -31,8 +30,8 @@ class Files(AndroidExtraction):
         # Run find command with correct args and parse results.
 
         # Check that full file printf options are suppported on first run.
-        if self.full_find == None:
+        if self.full_find is None:
-            output = self._adb_command(f"find '/' -maxdepth 1 -printf '%T@ %m %s %u %g %p\n' 2> /dev/null")
+            output = self._adb_command("find '/' -maxdepth 1 -printf '%T@ %m %s %u %g %p\n' 2> /dev/null")
             if not (output or output.strip().splitlines()):
                 # Full  find command failed to generate output, fallback to basic file arguments
                 self.full_find = False
@@ -40,7 +39,7 @@ class Files(AndroidExtraction):
                 self.full_find = True
 
         found_files = []
-        if self.full_find == True:
+        if self.full_find is True:
             # Run full file command and collect additonal file information.
             output = self._adb_command(f"find '{file_path}' -printf '%T@ %m %s %u %g %p\n' 2> /dev/null")
             for file_line in output.splitlines():
@@ -90,10 +89,6 @@ class Files(AndroidExtraction):
             return
 
         for result in self.results:
-            if self.indicators.check_filename(result["path"]):
-                self.log.warning("Found a known suspicous filename at path: \"%s\"", result["path"])
-                self.detected.append(result)
-
             if self.indicators.check_file_path(result["path"]):
                 self.log.warning("Found a known suspicous file at path: \"%s\"", result["path"])
                 self.detected.append(result)

mvt/android/modules/adb/packages.py

@@ -95,6 +95,9 @@ class Packages(AndroidExtraction):
         self._adb_connect()
 
         packages = self._adb_command("pm list packages -U -u -i -f")
+        if packages.strip() == "Error: Unknown option: -U":
+            packages = self._adb_command("pm list packages -u -i -f")
+
         for line in packages.split("\n"):
             line = line.strip()
             if not line.startswith("package:"):

mvt/common/module.py

@@ -30,7 +30,7 @@ class MVTModule(object):
     slug = None
 
     def __init__(self, file_path=None, base_folder=None, output_folder=None,
-                 fast_mode=False, log=None, results=[]):
+                 fast_mode=False, log=None, results=None):
         """Initialize module.
 
         :param file_path: Path to the module's database file, if there is any
@@ -51,7 +51,7 @@ class MVTModule(object):
         self.fast_mode = fast_mode
         self.log = log
         self.indicators = None
-        self.results = results
+        self.results = results if results else []
         self.detected = []
         self.timeline = []
         self.timeline_detected = []

mvt/common/version.py

@@ -6,7 +6,7 @@
 import requests
 from packaging import version
 
-MVT_VERSION = "1.3.2"
+MVT_VERSION = "1.4.7"
 
 
 def check_for_updates():

mvt/ios/cli.py

@@ -10,10 +10,10 @@ import click
 from rich.logging import RichHandler
 from rich.prompt import Prompt
 
-from mvt.common.help import HELP_MSG_MODULE, HELP_MSG_IOC
+from mvt.common.help import (HELP_MSG_FAST, HELP_MSG_IOC,
-from mvt.common.help import HELP_MSG_FAST, HELP_MSG_OUTPUT
+                             HELP_MSG_LIST_MODULES, HELP_MSG_MODULE,
-from mvt.common.help import HELP_MSG_LIST_MODULES
+                             HELP_MSG_OUTPUT)
-from mvt.common.indicators import Indicators, IndicatorsFileBadFormat
+from mvt.common.indicators import Indicators, download_indicators_files
 from mvt.common.logo import logo
 from mvt.common.module import run_module, save_timeline
 from mvt.common.options import MutuallyExclusiveOption
@@ -157,13 +157,7 @@ def check_backup(ctx, iocs, output, fast, backup_path, list_modules, module):
             ctx.exit(1)
 
     indicators = Indicators(log=log)
-    for ioc_path in iocs:
+    indicators.load_indicators_files(iocs)
-        try:
-            indicators.parse_stix2(ioc_path)
-        except IndicatorsFileBadFormat as e:
-            log.critical(e)
-            ctx.exit(1)
-    log.info("Loaded a total of %d indicators", indicators.ioc_count)
 
     timeline = []
     timeline_detected = []
@@ -174,8 +168,7 @@ def check_backup(ctx, iocs, output, fast, backup_path, list_modules, module):
         m = backup_module(base_folder=backup_path, output_folder=output, fast_mode=fast,
                           log=logging.getLogger(backup_module.__module__))
         m.is_backup = True
-
+        if indicators.total_ioc_count > 0:
-        if indicators.ioc_count > 0:
             m.indicators = indicators
             m.indicators.log = m.log
 
@@ -189,6 +182,10 @@ def check_backup(ctx, iocs, output, fast, backup_path, list_modules, module):
         if len(timeline_detected) > 0:
             save_timeline(timeline_detected, os.path.join(output, "timeline_detected.csv"))
 
+    if len(timeline_detected) > 0:
+        log.warning("The analysis of the backup produced %d detections!",
+                len(timeline_detected))
+
 
 #==============================================================================
 # Command: check-fs
@@ -220,13 +217,7 @@ def check_fs(ctx, iocs, output, fast, dump_path, list_modules, module):
             ctx.exit(1)
 
     indicators = Indicators(log=log)
-    for ioc_path in iocs:
+    indicators.load_indicators_files(iocs)
-        try:
-            indicators.parse_stix2(ioc_path)
-        except IndicatorsFileBadFormat as e:
-            log.critical(e)
-            ctx.exit(1)
-    log.info("Loaded a total of %d indicators", indicators.ioc_count)
 
     timeline = []
     timeline_detected = []
@@ -238,8 +229,7 @@ def check_fs(ctx, iocs, output, fast, dump_path, list_modules, module):
                       log=logging.getLogger(fs_module.__module__))
 
         m.is_fs_dump = True
-
+        if indicators.total_ioc_count > 0:
-        if iocs:
             m.indicators = indicators
             m.indicators.log = m.log
 
@@ -253,20 +243,23 @@ def check_fs(ctx, iocs, output, fast, dump_path, list_modules, module):
         if len(timeline_detected) > 0:
             save_timeline(timeline_detected, os.path.join(output, "timeline_detected.csv"))
 
+    if len(timeline_detected) > 0:
+        log.warning("The analysis of the filesystem produced %d detections!",
+                    len(timeline_detected))
 
 #==============================================================================
 # Command: check-iocs
 #==============================================================================
 @cli.command("check-iocs", help="Compare stored JSON results to provided indicators")
 @click.option("--iocs", "-i", type=click.Path(exists=True), multiple=True,
-              default=[], required=True, help=HELP_MSG_IOC)
+              default=[], help=HELP_MSG_IOC)
 @click.option("--list-modules", "-l", is_flag=True, help=HELP_MSG_LIST_MODULES)
 @click.option("--module", "-m", help=HELP_MSG_MODULE)
 @click.argument("FOLDER", type=click.Path(exists=True))
 @click.pass_context
 def check_iocs(ctx, iocs, list_modules, module, folder):
     all_modules = []
-    for entry in BACKUP_MODULES + FS_MODULES:
+    for entry in BACKUP_MODULES + FS_MODULES + MIXED_MODULES:
         if entry not in all_modules:
             all_modules.append(entry)
 
@@ -280,14 +273,9 @@ def check_iocs(ctx, iocs, list_modules, module, folder):
     log.info("Checking stored results against provided indicators...")
 
     indicators = Indicators(log=log)
-    for ioc_path in iocs:
+    indicators.load_indicators_files(iocs)
-        try:
-            indicators.parse_stix2(ioc_path)
-        except IndicatorsFileBadFormat as e:
-            log.critical(e)
-            ctx.exit(1)
-    log.info("Loaded a total of %d indicators", indicators.ioc_count)
 
+    total_detections = 0
     for file_name in os.listdir(folder):
         name_only, ext = os.path.splitext(file_name)
         file_path = os.path.join(folder, file_name)
@@ -304,11 +292,25 @@ def check_iocs(ctx, iocs, list_modules, module, folder):
 
             m = iocs_module.from_json(file_path,
                                       log=logging.getLogger(iocs_module.__module__))
-
+            if indicators.total_ioc_count > 0:
-            m.indicators = indicators
+                m.indicators = indicators
-            m.indicators.log = m.log
+                m.indicators.log = m.log
 
             try:
                 m.check_indicators()
             except NotImplementedError:
                 continue
+            else:
+                total_detections += len(m.detected)
+
+    if total_detections > 0:
+        log.warning("The check of the results produced %d detections!",
+                    total_detections)
+
+
+#==============================================================================
+# Command: download-iocs
+#==============================================================================
+@cli.command("download-iocs", help="Download public STIX2 indicators")
+def download_iocs():
+    download_indicators_files(log)

mvt/ios/modules/backup/configuration_profiles.py

@@ -7,6 +7,7 @@
 import os
 import plistlib
 from base64 import b64encode
+
 from mvt.common.utils import convert_timestamp_to_iso
 
 from ..base import IOSExtraction
@@ -45,8 +46,10 @@ class ConfigurationProfiles(IOSExtraction):
                 payload_content = result["plist"]["PayloadContent"][0]
 
                 # Alert on any known malicious configuration profiles in the indicator list.
-                if self.indicators.check_profile(result["plist"]["PayloadUUID"]):
+                ioc = self.indicators.check_profile(result["plist"]["PayloadUUID"])
+                if ioc:
                     self.log.warning(f"Found a known malicious configuration profile \"{result['plist']['PayloadDisplayName']}\" with UUID '{result['plist']['PayloadUUID']}'.")
+                    result["matched_indicator"] = ioc
                     self.detected.append(result)
                     continue
 
@@ -70,7 +73,7 @@ class ConfigurationProfiles(IOSExtraction):
             with open(conf_file_path, "rb") as handle:
                 try:
                     conf_plist = plistlib.load(handle)
-                except:
+                except Exception:
                     conf_plist = {}
 
             if "SignerCerts" in conf_plist:

mvt/ios/modules/backup/manifest.py

@@ -72,9 +72,7 @@ class Manifest(IOSExtraction):
             return
 
         for result in self.results:
-            if "relative_path" not in result:
+            if not result.get("relative_path"):
-                continue
-            if not result["relative_path"]:
                 continue
 
             if result["domain"]:
@@ -83,16 +81,15 @@ class Manifest(IOSExtraction):
                     self.detected.append(result)
                     continue
 
-            if self.indicators.check_filename(result["relative_path"]):
+            if self.indicators.check_file_path("/" + result["relative_path"]):
-                self.log.warning("Found a known malicious file at path: %s", result["relative_path"])
                 self.detected.append(result)
                 continue
 
-            relPath = result["relative_path"].lower()
+            rel_path = result["relative_path"].lower()
-            for ioc in self.indicators.ioc_domains:
+            for ioc in self.indicators.get_iocs("domains"):
-                if ioc.lower() in relPath:
+                if ioc["value"].lower() in rel_path:
                     self.log.warning("Found mention of domain \"%s\" in a backup file with path: %s",
-                                     ioc, relPath)
+                                     ioc["value"], rel_path)
                     self.detected.append(result)
 
     def run(self):

mvt/ios/modules/fs/analytics.py

@@ -37,20 +37,24 @@ class Analytics(IOSExtraction):
             return
 
         for result in self.results:
-            for ioc in self.indicators.ioc_processes:
+            for value in result.values():
-                for key in result.keys():
+                if not isinstance(value, str):
-                    if ioc == result[key]:
+                    continue
-                        self.log.warning("Found mention of a malicious process \"%s\" in %s file at %s",
+
-                                         ioc, result["artifact"], result["timestamp"])
+                ioc = self.indicators.check_process(value)
-                        self.detected.append(result)
+                if ioc:
-                        break
+                    self.log.warning("Found mention of a malicious process \"%s\" in %s file at %s",
-            for ioc in self.indicators.ioc_domains:
+                                     value, result["artifact"], result["timestamp"])
-                for key in result.keys():
+                    result["matched_indicator"] = ioc
-                    if ioc in str(result[key]):
+                    self.detected.append(result)
-                        self.log.warning("Found mention of a malicious domain \"%s\" in %s file at %s",
+                    continue
-                                         ioc, result["artifact"], result["timestamp"])
+
-                        self.detected.append(result)
+                ioc = self.indicators.check_domain(value)
-                        break
+                if ioc:
+                    self.log.warning("Found mention of a malicious domain \"%s\" in %s file at %s",
+                                     value, result["artifact"], result["timestamp"])
+                    result["matched_indicator"] = ioc
+                    self.detected.append(result)
 
     def _extract_analytics_data(self):
         artifact = self.file_path.split("/")[-1]
@@ -101,6 +105,7 @@ class Analytics(IOSExtraction):
                 timestamp = ""
                 data = plistlib.loads(row[1])
                 data["timestamp"] = timestamp
+
             data["artifact"] = artifact
 
             self.results.append(data)

mvt/ios/modules/fs/cache_files.py

@@ -34,13 +34,15 @@ class CacheFiles(IOSExtraction):
             return
 
         self.detected = {}
-        for key, items in self.results.items():
+        for key, values in self.results.items():
-            for item in items:
+            for value in values:
-                if self.indicators.check_domain(item["url"]):
+                ioc = self.indicators.check_domain(value["url"])
+                if ioc:
+                    value["matched_indicator"] = ioc
                     if key not in self.detected:
-                        self.detected[key] = [item, ]
+                        self.detected[key] = [value, ]
                     else:
-                        self.detected[key].append(item)
+                        self.detected[key].append(value)
 
     def _process_cache_file(self, file_path):
         self.log.info("Processing cache file at path: %s", file_path)

mvt/ios/modules/fs/filesystem.py

@@ -37,23 +37,25 @@ class Filesystem(IOSExtraction):
             return
 
         for result in self.results:
-            if self.indicators.check_file(result["path"]):
+            if "path" not in result:
-                self.log.warning("Found a known malicious file name at path: %s", result["path"])
+                continue
+
+            ioc = self.indicators.check_file_path(result["path"])
+            if ioc:
+                result["matched_indicator"] = ioc
                 self.detected.append(result)
 
-            if self.indicators.check_file_path(result["path"]):
+            # If we are instructed to run fast, we skip the rest.
-                self.log.warning("Found a known malicious file path at path: %s", result["path"])
-                self.detected.append(result)
-
-            # If we are instructed to run fast, we skip this.
             if self.fast_mode:
-                self.log.info("Flag --fast was enabled: skipping extended search for suspicious files/processes")
+                continue
-            else:
+
-                for ioc in self.indicators.ioc_processes:
+            for ioc in self.indicators.get_iocs("processes"):
-                    parts = result["path"].split("/")
+                parts = result["path"].split("/")
-                    if ioc in parts:
+                if ioc["value"] in parts:
-                        self.log.warning("Found a known malicious file/process at path: %s", result["path"])
+                    self.log.warning("Found known suspicious process name mentioned in file at path \"%s\" matching indicators from \"%s\"",
-                        self.detected.append(result)
+                                     result["path"], ioc["name"])
+                    result["matched_indicator"] = ioc
+                    self.detected.append(result)
 
     def run(self):
         for root, dirs, files in os.walk(self.base_folder):

mvt/ios/modules/fs/safari_favicon.py

@@ -37,7 +37,12 @@ class SafariFavicon(IOSExtraction):
             return
 
         for result in self.results:
-            if self.indicators.check_domain(result["url"]) or self.indicators.check_domain(result["icon_url"]):
+            ioc = self.indicators.check_domain(result["url"])
+            if not ioc:
+                ioc = self.indicators.check_domain(result["icon_url"])
+
+            if ioc:
+                result["matched_indicator"] = ioc
                 self.detected.append(result)
 
     def _process_favicon_db(self, file_path):

mvt/ios/modules/fs/shutdownlog.py

@@ -34,12 +34,20 @@ class ShutdownLog(IOSExtraction):
             return
 
         for result in self.results:
-            for ioc in self.indicators.ioc_processes:
+            ioc = self.indicators.check_file_path(result["client"])
+            if ioc:
+                result["matched_indicator"] = ioc
+                self.detected.append(result)
+                continue
+
+            for ioc in self.indicators.get_iocs("processes"):
                 parts = result["client"].split("/")
                 if ioc in parts:
                     self.log.warning("Found mention of a known malicious process \"%s\" in shutdown.log",
                                      ioc)
+                    result["matched_indicator"] = ioc
                     self.detected.append(result)
+                    continue
 
     def process_shutdownlog(self, content):
         current_processes = []

mvt/ios/modules/fs/webkit_base.py

@@ -18,9 +18,11 @@ class WebkitBase(IOSExtraction):
         if not self.indicators:
             return
 
-        for item in self.results:
+        for result in self.results:
-            if self.indicators.check_domain(item["url"]):
+            ioc = self.indicators.check_domain(result["url"])
-                self.detected.append(item)
+            if ioc:
+                result["matched_indicator"] = ioc
+                self.detected.append(result)
 
     def _process_webkit_folder(self, root_paths):
         for found_path in self._get_fs_files_from_patterns(root_paths):

mvt/ios/modules/mixed/__init__.py

@@ -16,13 +16,13 @@ from .net_datausage import Datausage
 from .osanalytics_addaily import OSAnalyticsADDaily
 from .safari_browserstate import SafariBrowserState
 from .safari_history import SafariHistory
+from .shortcuts import Shortcuts
 from .sms import SMS
 from .sms_attachments import SMSAttachments
 from .tcc import TCC
 from .webkit_resource_load_statistics import WebkitResourceLoadStatistics
 from .webkit_session_resource_log import WebkitSessionResourceLog
 from .whatsapp import Whatsapp
-from .shortcuts import Shortcuts
 
 MIXED_MODULES = [Calls, ChromeFavicon, ChromeHistory, Contacts, FirefoxFavicon,
                  FirefoxHistory, IDStatusCache, InteractionC, LocationdClients,

mvt/ios/modules/mixed/chrome_favicon.py

@@ -42,7 +42,12 @@ class ChromeFavicon(IOSExtraction):
             return
 
         for result in self.results:
-            if self.indicators.check_domain(result["url"]) or self.indicators.check_domain(result["icon_url"]):
+            ioc = self.indicators.check_domain(result["url"])
+            if not ioc:
+                ioc = self.indicators.check_domain(result["icon_url"])
+
+            if ioc:
+                result["matched_indicator"] = ioc
                 self.detected.append(result)
 
     def run(self):

mvt/ios/modules/mixed/chrome_history.py

@@ -41,7 +41,9 @@ class ChromeHistory(IOSExtraction):
             return
 
         for result in self.results:
-            if self.indicators.check_domain(result["url"]):
+            ioc = self.indicators.check_domain(result["url"])
+            if ioc:
+                result["matched_indicator"] = ioc
                 self.detected.append(result)
 
     def run(self):

mvt/ios/modules/mixed/firefox_favicon.py

@@ -40,8 +40,12 @@ class FirefoxFavicon(IOSExtraction):
             return
 
         for result in self.results:
-            if (self.indicators.check_domain(result.get("url", "")) or
+            ioc = self.indicators.check_domain(result.get("url", ""))
-                    self.indicators.check_domain(result.get("history_url", ""))):
+            if not ioc:
+                ioc = self.indicators.check_domain(result.get("history_url", ""))
+
+            if ioc:
+                result["matched_indicator"] = ioc
                 self.detected.append(result)
 
     def run(self):

mvt/ios/modules/mixed/firefox_history.py

@@ -44,7 +44,9 @@ class FirefoxHistory(IOSExtraction):
             return
 
         for result in self.results:
-            if self.indicators.check_domain(result["url"]):
+            ioc = self.indicators.check_domain(result["url"])
+            if ioc:
+                result["matched_indicator"] = ioc
                 self.detected.append(result)
 
     def run(self):

mvt/ios/modules/mixed/idstatuscache.py

@@ -43,7 +43,9 @@ class IDStatusCache(IOSExtraction):
         for result in self.results:
             if result.get("user", "").startswith("mailto:"):
                 email = result["user"][7:].strip("'")
-                if self.indicators.check_email(email):
+                ioc = self.indicators.check_email(email)
+                if ioc:
+                    result["matched_indicator"] = ioc
                     self.detected.append(result)
                     continue
 

mvt/ios/modules/mixed/locationd.py

@@ -41,13 +41,13 @@ class LocationdClients(IOSExtraction):
 
     def serialize(self, record):
         records = []
-        for ts in self.timestamps:
+        for timestamp in self.timestamps:
-            if ts in record.keys():
+            if timestamp in record.keys():
                 records.append({
-                    "timestamp": record[ts],
+                    "timestamp": record[timestamp],
                     "module": self.__class__.__name__,
-                    "event": ts,
+                    "event": timestamp,
-                    "data": f"{ts} from {record['package']}"
+                    "data": f"{timestamp} from {record['package']}"
                 })
 
         return records
@@ -60,8 +60,40 @@ class LocationdClients(IOSExtraction):
             parts = result["package"].split("/")
             proc_name = parts[len(parts)-1]
 
-            if self.indicators.check_process(proc_name):
+            ioc = self.indicators.check_process(proc_name)
+            if ioc:
+                self.log.warning("Found a suspicious process name in LocationD entry %s",
+                                 result["package"])
+                result["matched_indicator"] = ioc
                 self.detected.append(result)
+                continue
+
+            if "BundlePath" in result:
+                ioc = self.indicators.check_file_path(result["BundlePath"])
+                if ioc:
+                    self.log.warning("Found a suspicious file path in Location D: %s",
+                                     result["BundlePath"])
+                    result["matched_indicator"] = ioc
+                    self.detected.append(result)
+                    continue
+
+            if "Executable" in result:
+                ioc = self.indicators.check_file_path(result["Executable"])
+                if ioc:
+                    self.log.warning("Found a suspicious file path in Location D: %s",
+                                     result["Executable"])
+                    result["matched_indicator"] = ioc
+                    self.detected.append(result)
+                    continue
+
+            if "Registered" in result:
+                ioc = self.indicators.check_file_path(result["Registered"])
+                if ioc:
+                    self.log.warning("Found a suspicious file path in Location D: %s",
+                                     result["Registered"])
+                    result["matched_indicator"] = ioc
+                    self.detected.append(result)
+                    continue
 
     def _extract_locationd_entries(self, file_path):
         with open(file_path, "rb") as handle:

mvt/ios/modules/mixed/osanalytics_addaily.py

@@ -41,7 +41,9 @@ class OSAnalyticsADDaily(IOSExtraction):
             return
 
         for result in self.results:
-            if self.indicators.check_process(result["package"]):
+            ioc = self.indicators.check_process(result["package"])
+            if ioc:
+                result["matched_indicator"] = ioc
                 self.detected.append(result)
 
     def run(self):

mvt/ios/modules/mixed/safari_browserstate.py

@@ -44,16 +44,22 @@ class SafariBrowserState(IOSExtraction):
             return
 
         for result in self.results:
-            if "tab_url" in result and self.indicators.check_domain(result["tab_url"]):
+            if "tab_url" in result:
-                self.detected.append(result)
+                ioc = self.indicators.check_domain(result["tab_url"])
-                continue
+                if ioc:
+                    result["matched_indicator"] = ioc
+                    self.detected.append(result)
+                    continue
 
             if "session_data" not in result:
                 continue
 
             for session_entry in result["session_data"]:
-                if "entry_url" in session_entry and self.indicators.check_domain(session_entry["entry_url"]):
+                if "entry_url" in session_entry:
-                    self.detected.append(result)
+                    ioc = self.indicators.check_domain(session_entry["entry_url"])
+                    if ioc:
+                        result["matched_indicator"] = ioc
+                        self.detected.append(result)
 
     def _process_browser_state_db(self, db_path):
         conn = sqlite3.connect(db_path)

mvt/ios/modules/mixed/safari_history.py

@@ -80,7 +80,9 @@ class SafariHistory(IOSExtraction):
             return
 
         for result in self.results:
-            if self.indicators.check_domain(result["url"]):
+            ioc = self.indicators.check_domain(result["url"])
+            if ioc:
+                result["matched_indicator"] = ioc
                 self.detected.append(result)
 
     def _process_history_db(self, history_path):

mvt/ios/modules/mixed/shortcuts.py

@@ -3,12 +3,13 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
-import sqlite3
 import io
-import plistlib
 import itertools
+import plistlib
+import sqlite3
 
-from mvt.common.utils import check_for_links, convert_mactime_to_unix, convert_timestamp_to_iso
+from mvt.common.utils import (check_for_links, convert_mactime_to_unix,
+                              convert_timestamp_to_iso)
 
 from ..base import IOSExtraction
 
@@ -33,21 +34,31 @@ class Shortcuts(IOSExtraction):
         found_urls = ""
         if record["action_urls"]:
             found_urls = "- URLs in actions: {}".format(", ".join(record["action_urls"]))
+        desc = ""
+        if record["description"]:
+            desc = record["description"].decode('utf-8', errors='ignore')
 
-        return {
+        return [{
             "timestamp": record["isodate"],
             "module": self.__class__.__name__,
-            "event": "shortcut",
+            "event": "shortcut_created",
-            "data": f"iOS Shortcut '{record['shortcut_name']}': {record['description']} {found_urls}"
+            "data": f"iOS Shortcut '{record['shortcut_name'].decode('utf-8')}': {desc} {found_urls}"
-        }
+        }, {
+            "timestamp": record["modified_date"],
+            "module": self.__class__.__name__,
+            "event": "shortcut_modified",
+            "data": f"iOS Shortcut '{record['shortcut_name'].decode('utf-8')}': {desc} {found_urls}"
+        }]
 
     def check_indicators(self):
         if not self.indicators:
             return
 
-        for action in self.results:
+        for result in self.results:
-            if self.indicators.check_domains(action["action_urls"]):
+            ioc = self.indicators.check_domains(result["action_urls"])
-                self.detected.append(action)
+            if ioc:
+                result["matched_indicator"] = ioc
+                self.detected.append(result)
 
     def run(self):
         self._find_ios_database(backup_ids=SHORTCUT_BACKUP_IDS,
@@ -57,17 +68,25 @@ class Shortcuts(IOSExtraction):
         conn = sqlite3.connect(self.file_path)
         conn.text_factory = bytes
         cur = conn.cursor()
-        cur.execute("""
+        try:
-              SELECT
+            cur.execute("""
-                  ZSHORTCUT.Z_PK as "shortcut_id",
+                SELECT
-                  ZSHORTCUT.ZNAME as "shortcut_name",
+                    ZSHORTCUT.Z_PK as "shortcut_id",
-                  ZSHORTCUT.ZCREATIONDATE as "created_date",
+                    ZSHORTCUT.ZNAME as "shortcut_name",
-                  ZSHORTCUT.ZMODIFICATIONDATE as "modified_date",
+                    ZSHORTCUT.ZCREATIONDATE as "created_date",
-                  ZSHORTCUT.ZACTIONSDESCRIPTION as "description",
+                    ZSHORTCUT.ZMODIFICATIONDATE as "modified_date",
-                  ZSHORTCUTACTIONS.ZDATA as "action_data"
+                    ZSHORTCUT.ZACTIONSDESCRIPTION as "description",
-              FROM ZSHORTCUT
+                    ZSHORTCUTACTIONS.ZDATA as "action_data"
-              LEFT JOIN ZSHORTCUTACTIONS ON ZSHORTCUTACTIONS.ZSHORTCUT == ZSHORTCUT.Z_PK;
+                FROM ZSHORTCUT
-        """)
+                LEFT JOIN ZSHORTCUTACTIONS ON ZSHORTCUTACTIONS.ZSHORTCUT == ZSHORTCUT.Z_PK;
+            """)
+        except sqlite3.OperationalError:
+            # Table ZSHORTCUT does not exist
+            self.log.info("Invalid shortcut database format, skipping...")
+            cur.close()
+            conn.close()
+            return
+
         names = [description[0] for description in cur.description]
 
         for item in cur:
@@ -83,14 +102,13 @@ class Shortcuts(IOSExtraction):
                 action["identifier"] = action_entry["WFWorkflowActionIdentifier"]
                 action["parameters"] = action_entry["WFWorkflowActionParameters"]
 
-                # URLs might be in multiple fields, do a simple regex search across the parameters
+                # URLs might be in multiple fields, do a simple regex search across the parameters.
                 extracted_urls = check_for_links(str(action["parameters"]))
 
-                # Remove quoting characters that may have been captured by the regex
+                # Remove quoting characters that may have been captured by the regex.
                 action["urls"] = [url.rstrip("',") for url in extracted_urls]
                 actions.append(action)
 
-            # pprint.pprint(actions)
             shortcut["isodate"] = convert_timestamp_to_iso(convert_mactime_to_unix(shortcut.pop("created_date")))
             shortcut["modified_date"] = convert_timestamp_to_iso(convert_mactime_to_unix(shortcut["modified_date"]))
             shortcut["parsed_actions"] = len(actions)

mvt/ios/modules/mixed/sms.py

@@ -41,10 +41,12 @@ class SMS(IOSExtraction):
         if not self.indicators:
             return
 
-        for message in self.results:
+        for result in self.results:
-            message_links = check_for_links(message.get("text", ""))
+            message_links = check_for_links(result.get("text", ""))
-            if self.indicators.check_domains(message_links):
+            ioc = self.indicators.check_domains(message_links)
-                self.detected.append(message)
+            if ioc:
+                result["matched_indicator"] = ioc
+                self.detected.append(result)
 
     def run(self):
         self._find_ios_database(backup_ids=SMS_BACKUP_IDS,