Bumped version

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
-    #   run: |
-    #     pytest
+    - name: Test with pytest
+      run: pytest

docs/iocs.md

@@ -41,6 +41,6 @@ export MVT_STIX2="/home/user/IOC1.stix2:/home/user/IOC2.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://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`.
+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

@@ -139,7 +139,7 @@ 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.ioc_count:
+        if indicators.total_ioc_count:
             m.indicators = indicators
             m.indicators.log = m.log
         if serial:
@@ -190,7 +190,7 @@ 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.ioc_count:
+        if indicators.total_ioc_count:
             m.indicators = indicators
             m.indicators.log = m.log
         if serial:

mvt/android/modules/adb/files.py

@@ -89,10 +89,6 @@ class Files(AndroidExtraction):
             return
 
         for result in self.results:
-            if self.indicators.check_file_name(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/indicators.py

@@ -21,19 +21,8 @@ class Indicators:
     def __init__(self, log=None):
         self.data_dir = user_data_dir("mvt")
         self.log = log
-        self.ioc_domains = []
-        self.ioc_processes = []
-        self.ioc_emails = []
         self.ioc_files = []
-        self.ioc_files_sha256 = []
-        self.ioc_app_ids = []
-        self.ios_profile_ids = []
-        self.ioc_count = 0
-
-    def _add_indicator(self, ioc, iocs_list):
-        if ioc not in iocs_list:
-            iocs_list.append(ioc)
-            self.ioc_count += 1
+        self.total_ioc_count = 0
 
     def _load_downloaded_indicators(self):
         if not os.path.isdir(self.data_dir):
@@ -45,7 +34,7 @@ class Indicators:
 
     def _check_stix2_env_variable(self):
         """
-        Checks if a variable MVT_STIX2 contains path to STIX Files
+        Checks if a variable MVT_STIX2 contains path to STIX Files.
         """
         if "MVT_STIX2" not in os.environ:
             return False
@@ -57,20 +46,28 @@ class Indicators:
             else:
                 self.log.info("Invalid STIX2 path %s in MVT_STIX2 environment variable", path)
 
-    def load_indicators_files(self, files):
-        """
-        Load a list of indicators files
-        """
-        for file_path in files:
-            if os.path.isfile(file_path):
-                self.parse_stix2(file_path)
-            else:
-                self.log.warning("This indicators file %s does not exist", file_path)
+    def _generate_indicators_file(self):
+        return {
+            "name": "",
+            "description": "",
+            "file_name": "",
+            "file_path": "",
+            "domains": [],
+            "processes": [],
+            "emails": [],
+            "file_names": [],
+            "file_paths": [],
+            "files_sha256": [],
+            "app_ids": [],
+            "ios_profile_ids": [],
+            "count": 0,
+        }
 
-        # Load downloaded indicators and any indicators from env variable
-        self._load_downloaded_indicators()
-        self._check_stix2_env_variable()
-        self.log.info("Loaded a total of %d unique indicators", self.ioc_count)
+    def _add_indicator(self, ioc, ioc_file, iocs_list):
+        if ioc not in iocs_list:
+            iocs_list.append(ioc)
+            ioc_file["count"] += 1
+            self.total_ioc_count += 1
 
     def parse_stix2(self, file_path):
         """Extract indicators from a STIX2 file.
@@ -80,6 +77,11 @@ class Indicators:
 
         """
         self.log.info("Parsing STIX2 indicators file at path %s", file_path)
+
+        ioc_file = self._generate_indicators_file()
+        ioc_file["file_path"] = file_path
+        ioc_file["file_name"] = os.path.basename(file_path)
+
         with open(file_path, "r") as handle:
             try:
                 data = json.load(handle)
@@ -88,7 +90,13 @@ class Indicators:
                 return
 
         for entry in data.get("objects", []):
-            if entry.get("type", "") != "indicator":
+            entry_type = entry.get("type", "")
+            if entry_type == "malware":
+                ioc_file["name"] = entry.get("name", "") or ioc_file["file_name"]
+                ioc_file["description"] = entry.get("description", "") or ioc_file["file_name"]
+                continue
+
+            if entry_type != "indicator":
                 continue
 
             key, value = entry.get("pattern", "").strip("[]").split("=")
@@ -97,26 +105,67 @@ class Indicators:
             if key == "domain-name:value":
                 # We force domain names to lower case.
                 self._add_indicator(ioc=value.lower(),
-                                    iocs_list=self.ioc_domains)
+                                    ioc_file=ioc_file,
+                                    iocs_list=ioc_file["domains"])
             elif key == "process:name":
                 self._add_indicator(ioc=value,
-                                    iocs_list=self.ioc_processes)
+                                    ioc_file=ioc_file,
+                                    iocs_list=ioc_file["processes"])
             elif key == "email-addr:value":
                 # We force email addresses to lower case.
                 self._add_indicator(ioc=value.lower(),
-                                    iocs_list=self.ioc_emails)
+                                    ioc_file=ioc_file,
+                                    iocs_list=ioc_file["emails"])
             elif key == "file:name":
                 self._add_indicator(ioc=value,
-                                    iocs_list=self.ioc_files)
-            elif key == "app:id":
+                                    ioc_file=ioc_file,
+                                    iocs_list=ioc_file["file_names"])
+            elif key == "file:path":
                 self._add_indicator(ioc=value,
-                                    iocs_list=self.ioc_app_ids)
-            elif key == "configuration-profile:id":
-                self._add_indicator(ioc=value,
-                                    iocs_list=self.ios_profile_ids)
+                                    ioc_file=ioc_file,
+                                    iocs_list=ioc_file["file_paths"])
             elif key == "file:hashes.sha256":
                 self._add_indicator(ioc=value,
-                                    iocs_list=self.ioc_files_sha256)
+                                    ioc_file=ioc_file,
+                                    iocs_list=ioc_file["files_sha256"])
+            elif key == "app:id":
+                self._add_indicator(ioc=value,
+                                    ioc_file=ioc_file,
+                                    iocs_list=ioc_file["app_ids"])
+            elif key == "configuration-profile:id":
+                self._add_indicator(ioc=value,
+                                    ioc_file=ioc_file,
+                                    iocs_list=ioc_file["ios_profile_ids"])
+
+        self.log.info("Loaded %d indicators from \"%s\" indicators file",
+                      ioc_file["count"], ioc_file["name"])
+
+        self.ioc_files.append(ioc_file)
+
+    def load_indicators_files(self, files, load_default=True):
+        """
+        Load a list of indicators files.
+        """
+        for file_path in files:
+            if os.path.isfile(file_path):
+                self.parse_stix2(file_path)
+            else:
+                self.log.warning("This indicators file %s does not exist", file_path)
+
+        # Load downloaded indicators and any indicators from env variable.
+        if load_default:
+            self._load_downloaded_indicators()
+
+        self._check_stix2_env_variable()
+        self.log.info("Loaded a total of %d unique indicators", self.total_ioc_count)
+
+    def get_iocs(self, ioc_type):
+        for ioc_file in self.ioc_files:
+            for ioc in ioc_file.get(ioc_type, []):
+                yield {
+                    "value": ioc,
+                    "name": ioc_file["name"]
+                }
 
     def check_domain(self, url) -> bool:
         """Check if a given URL matches any of the provided domain indicators.
@@ -128,7 +177,7 @@ class Indicators:
 
         """
         # TODO: If the IOC domain contains a subdomain, it is not currently
-        # being matched.
+        #       being matched.
         if not url:
             return False
 
@@ -158,33 +207,36 @@ class Indicators:
         except Exception:
             # If URL parsing failed, we just try to do a simple substring
             # match.
-            for ioc in self.ioc_domains:
-                if ioc.lower() in url:
-                    self.log.warning("Maybe found a known suspicious domain: %s", url)
+            for ioc in self.get_iocs("domains"):
+                if ioc["value"].lower() in url:
+                    self.log.warning("Maybe found a known suspicious domain %s matching indicators from \"%s\"",
+                                     url, ioc["name"])
                     return True
 
             # If nothing matched, we can quit here.
             return False
 
         # If all parsing worked, we start walking through available domain indicators.
-        for ioc in self.ioc_domains:
+        for ioc in self.get_iocs("domains"):
             # First we check the full domain.
-            if final_url.domain.lower() == ioc:
+            if final_url.domain.lower() == ioc["value"]:
                 if orig_url.is_shortened and orig_url.url != final_url.url:
-                    self.log.warning("Found a known suspicious domain %s shortened as %s",
-                                     final_url.url, orig_url.url)
+                    self.log.warning("Found a known suspicious domain %s shortened as %s matching indicators from \"%s\"",
+                                     final_url.url, orig_url.url, ioc["name"])
                 else:
-                    self.log.warning("Found a known suspicious domain: %s", final_url.url)
+                    self.log.warning("Found a known suspicious domain %s matching indicators from \"%s\"",
+                                     final_url.url, ioc["name"])
 
                 return True
 
             # Then we just check the top level domain.
-            if final_url.top_level.lower() == ioc:
+            if final_url.top_level.lower() == ioc["value"]:
                 if orig_url.is_shortened and orig_url.url != final_url.url:
-                    self.log.warning("Found a sub-domain matching a known suspicious top level %s shortened as %s",
-                                     final_url.url, orig_url.url)
+                    self.log.warning("Found a sub-domain with suspicious top level %s shortened as %s matching indicators from \"%s\"",
+                                     final_url.url, orig_url.url, ioc["name"])
                 else:
-                    self.log.warning("Found a sub-domain matching a known suspicious top level: %s", final_url.url)
+                    self.log.warning("Found a sub-domain with a suspicious top level %s matching indicators from \"%s\"",
+                                     final_url.url, ioc["name"])
 
                 return True
 
@@ -222,14 +274,16 @@ class Indicators:
             return False
 
         proc_name = os.path.basename(process)
-        if proc_name in self.ioc_processes:
-            self.log.warning("Found a known suspicious process name \"%s\"", process)
-            return True
+        for ioc in self.get_iocs("processes"):
+            if proc_name == ioc["value"]:
+                self.log.warning("Found a known suspicious process name \"%s\" matching indicators from \"%s\"",
+                                 process, ioc["name"])
+                return True
 
-        if len(proc_name) == 16:
-            for bad_proc in self.ioc_processes:
-                if bad_proc.startswith(proc_name):
-                    self.log.warning("Found a truncated known suspicious process name \"%s\"", process)
+            if len(proc_name) == 16:
+                if ioc["value"].startswith(proc_name):
+                    self.log.warning("Found a truncated known suspicious process name \"%s\" matching indicators from \"%s\"",
+                                     process, ioc["name"])
                     return True
 
         return False
@@ -265,36 +319,37 @@ class Indicators:
         if not email:
             return False
 
-        if email.lower() in self.ioc_emails:
-            self.log.warning("Found a known suspicious email address: \"%s\"", email)
-            return True
+        for ioc in self.get_iocs("emails"):
+            if email.lower() == ioc["value"].lower():
+                self.log.warning("Found a known suspicious email address \"%s\" matching indicators from \"%s\"",
+                                 email, ioc["name"])
+                return True
 
         return False
 
-    def check_file_name(self, file_path) -> bool:
-        """Check the provided file path against the list of file indicators.
+    def check_file_name(self, file_name) -> bool:
+        """Check the provided file name against the list of file indicators.
 
-        :param file_path: File path or file name to check against file
+        :param file_name: File name to check against file
         indicators
-        :type file_path: str
-        :returns: True if the file path matched an indicator, otherwise False
+        :type file_name: str
+        :returns: True if the file name matched an indicator, otherwise False
         :rtype: bool
 
         """
-        if not file_path:
+        if not file_name:
             return False
 
-        file_name = os.path.basename(file_path)
-        if file_name in self.ioc_files:
-            return True
+        for ioc in self.get_iocs("file_names"):
+            if ioc["value"] == file_name:
+                self.log.warning("Found a known suspicious file name \"%s\" matching indicators from \"%s\"",
+                                 file_name, ioc["name"])
+                return True
 
         return False
 
-    # TODO: The difference between check_file_name() and check_file_path()
-    #       needs to be more explicit and clear. Probably, the two should just
-    #       be combined into one function.
     def check_file_path(self, file_path) -> bool:
-        """Check the provided file path against the list of file indicators.
+        """Check the provided file path against the list of file indicators (both path and name).
 
         :param file_path: File path or file name to check against file
         indicators
@@ -306,9 +361,14 @@ class Indicators:
         if not file_path:
             return False
 
-        for ioc_file in self.ioc_files:
+        if self.check_file_name(os.path.basename(file_path)):
+            return True
+
+        for ioc in self.get_iocs("file_paths"):
             # Strip any trailing slash from indicator paths to match directories.
-            if file_path.startswith(ioc_file.rstrip("/")):
+            if file_path.startswith(ioc["value"].rstrip("/")):
+                self.log.warning("Found a known suspicious file path \"%s\" matching indicators form \"%s\"",
+                                 file_path, ioc["name"])
                 return True
 
         return False
@@ -322,15 +382,18 @@ class Indicators:
         :rtype: bool
 
         """
-        if profile_uuid in self.ios_profile_ids:
-            return True
+        for ioc in self.get_iocs("ios_profile_ids"):
+            if profile_uuid in ioc["value"]:
+                self.log.warning("Found a known suspicious profile ID \"%s\" matching indicators from \"%s\"",
+                                 profile_uuid, ioc["name"])
+                return True
 
         return False
 
 
 def download_indicators_files(log):
     """
-    Download indicators from repo into MVT app data directory
+    Download indicators from repo into MVT app data directory.
     """
     data_dir = user_data_dir("mvt")
     if not os.path.isdir(data_dir):
@@ -357,4 +420,5 @@ def download_indicators_files(log):
         # Write file to disk. This will overwrite any older version of the STIX2 file.
         with io.open(ioc_path, "w") as f:
             f.write(res.text)
+
         log.info("Saved indicator file to '%s'", os.path.basename(ioc_path))

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.4.2"
+MVT_VERSION = "1.4.6"
 
 
 def check_for_updates():

mvt/ios/cli.py

@@ -168,7 +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.ioc_count:
+        if indicators.total_ioc_count > 0:
             m.indicators = indicators
             m.indicators.log = m.log
 
@@ -182,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
@@ -225,7 +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.ioc_count:
+        if indicators.total_ioc_count > 0:
             m.indicators = indicators
             m.indicators.log = m.log
 
@@ -239,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)
 
@@ -268,6 +275,7 @@ def check_iocs(ctx, iocs, list_modules, module, folder):
     indicators = Indicators(log=log)
     indicators.load_indicators_files(iocs)
 
+    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)
@@ -284,7 +292,7 @@ def check_iocs(ctx, iocs, list_modules, module, folder):
 
             m = iocs_module.from_json(file_path,
                                       log=logging.getLogger(iocs_module.__module__))
-            if indicators.ioc_count:
+            if indicators.total_ioc_count > 0:
                 m.indicators = indicators
                 m.indicators.log = m.log
 
@@ -292,11 +300,17 @@ def check_iocs(ctx, iocs, list_modules, module, folder):
                 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_indicators():
+def download_iocs():
     download_indicators_files(log)

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:
-                continue
-            if not result["relative_path"]:
+            if not result.get("relative_path"):
                 continue
 
             if result["domain"]:
@@ -83,16 +81,15 @@ class Manifest(IOSExtraction):
                     self.detected.append(result)
                     continue
 
-            if self.indicators.check_file_name(result["relative_path"]):
-                self.log.warning("Found a known malicious file at path: %s", result["relative_path"])
+            if self.indicators.check_file_path("/" + result["relative_path"]):
                 self.detected.append(result)
                 continue
 
-            relPath = result["relative_path"].lower()
-            for ioc in self.indicators.ioc_domains:
-                if ioc.lower() in relPath:
+            rel_path = result["relative_path"].lower()
+            for ioc in self.indicators.get_iocs("domains"):
+                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,20 @@ class Analytics(IOSExtraction):
             return
 
         for result in self.results:
-            for ioc in self.indicators.ioc_processes:
-                for key in result.keys():
-                    if ioc == result[key]:
-                        self.log.warning("Found mention of a malicious process \"%s\" in %s file at %s",
-                                         ioc, result["artifact"], result["timestamp"])
-                        self.detected.append(result)
-                        break
-            for ioc in self.indicators.ioc_domains:
-                for key in result.keys():
-                    if ioc in str(result[key]):
-                        self.log.warning("Found mention of a malicious domain \"%s\" in %s file at %s",
-                                         ioc, result["artifact"], result["timestamp"])
-                        self.detected.append(result)
-                        break
+            for value in result.values():
+                if not isinstance(value, str):
+                    continue
+
+                if self.indicators.check_process(value):
+                    self.log.warning("Found mention of a malicious process \"%s\" in %s file at %s",
+                                     value, result["artifact"], result["timestamp"])
+                    self.detected.append(result)
+                    continue
+
+                if self.indicators.check_domain(value):
+                    self.log.warning("Found mention of a malicious domain \"%s\" in %s file at %s",
+                                     value, result["artifact"], result["timestamp"])
+                    self.detected.append(result)
 
     def _extract_analytics_data(self):
         artifact = self.file_path.split("/")[-1]
@@ -101,6 +101,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,13 @@ class CacheFiles(IOSExtraction):
             return
 
         self.detected = {}
-        for key, items in self.results.items():
-            for item in items:
-                if self.indicators.check_domain(item["url"]):
+        for key, values in self.results.items():
+            for value in values:
+                if self.indicators.check_domain(value["url"]):
                     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,22 @@ class Filesystem(IOSExtraction):
             return
 
         for result in self.results:
-            if self.indicators.check_file(result["path"]):
-                self.log.warning("Found a known malicious file name at path: %s", result["path"])
-                self.detected.append(result)
+            if "path" not in result:
+                continue
 
             if self.indicators.check_file_path(result["path"]):
-                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 we are instructed to run fast, we skip the rest.
             if self.fast_mode:
-                self.log.info("Flag --fast was enabled: skipping extended search for suspicious files/processes")
-            else:
-                for ioc in self.indicators.ioc_processes:
-                    parts = result["path"].split("/")
-                    if ioc in parts:
-                        self.log.warning("Found a known malicious file/process at path: %s", result["path"])
-                        self.detected.append(result)
+                continue
+
+            for ioc in self.indicators.get_iocs("processes"):
+                parts = result["path"].split("/")
+                if ioc["value"] in parts:
+                    self.log.warning("Found known suspicious process name mentioned in file at path \"%s\" matching indicators from \"%s\"",
+                                     result["path"], ioc["name"])
+                    self.detected.append(result)
 
     def run(self):
         for root, dirs, files in os.walk(self.base_folder):

mvt/ios/modules/fs/shutdownlog.py

@@ -34,12 +34,17 @@ class ShutdownLog(IOSExtraction):
             return
 
         for result in self.results:
-            for ioc in self.indicators.ioc_processes:
+            if self.indicators.check_file_path(result["client"]):
+                    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)
                     self.detected.append(result)
+                    continue
 
     def process_shutdownlog(self, content):
         current_processes = []

mvt/ios/modules/mixed/locationd.py

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

mvt/ios/modules/mixed/shortcuts.py

@@ -34,13 +34,21 @@ 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",
-            "data": f"iOS Shortcut '{record['shortcut_name']}': {record['description']} {found_urls}"
-        }
+            "event": "shortcut_created",
+            "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:
@@ -92,14 +100,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/tcc.py

@@ -66,6 +66,15 @@ class TCC(IOSExtraction):
                 "data": msg
             }
 
+    def check_indicators(self):
+        if not self.indicators:
+            return
+
+        for result in self.results:
+            if self.indicators.check_process(result["client"]):
+                self.log.warning("Found malicious process in TCC database: %s", result["client"])
+                self.detected.append(result)
+
     def process_db(self, file_path):
         conn = sqlite3.connect(file_path)
         cur = conn.cursor()

mvt/ios/modules/mixed/whatsapp.py

@@ -35,6 +35,7 @@ class Whatsapp(IOSExtraction):
         links_text = ""
         if record["links"]:
             links_text = " - Embedded links: " + ", ".join(record["links"])
+
         return {
             "timestamp": record.get("isodate"),
             "module": self.__class__.__name__,
@@ -47,7 +48,7 @@ class Whatsapp(IOSExtraction):
             return
 
         for message in self.results:
-            if self.indicators.check_domains(message["links"]):
+            if self.indicators.check_domains(message.get("links", [])):
                 self.detected.append(message)
 
     def run(self):
@@ -83,14 +84,15 @@ class Whatsapp(IOSExtraction):
             message["isodate"] = convert_timestamp_to_iso(convert_mactime_to_unix(message.get("ZMESSAGEDATE")))
             message["ZTEXT"] = message["ZTEXT"] if message["ZTEXT"] else ""
 
-            # Extract links from the WhatsApp message. URLs can be stored in multiple fields/columns. Check each of them!
+            # Extract links from the WhatsApp message. URLs can be stored in multiple fields/columns.
+            # Check each of them!
             message_links = []
             fields_with_links = ["ZTEXT", "ZMATCHEDTEXT", "ZMEDIAURL", "ZCONTENT1", "ZCONTENT2"]
             for field in fields_with_links:
                 if message.get(field):
                     message_links.extend(check_for_links(message.get(field, "")))
 
-            # Remove WhatsApp internal media URLs
+            # Remove WhatsApp internal media URLs.
             filtered_links = []
             for link in message_links:
                 if not (link.startswith("https://mmg-fna.whatsapp.net/") or link.startswith("https://mmg.whatsapp.net/")):

mvt/ios/versions.py

@@ -234,6 +234,8 @@ IPHONE_IOS_VERSIONS = [
     {"build": "19A404", "version": "15.0.2"},
     {"build": "19B74", "version": "15.1"},
     {"build": "19B81", "version": "15.1.1"},
+    {"build": "19C56", "version": "15.2"},
+    {"build": "19C63", "version": "15.2.1"},
 ]
 
 

tests/artifacts/.gitignore

@@ -0,0 +1 @@
+test.stix2

tests/artifacts/generate_stix.py

@@ -0,0 +1,50 @@
+# 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 os
+
+from stix2.v21 import Bundle, Indicator, Malware, Relationship
+
+
+def generate_test_stix_file(file_path):
+    if os.path.isfile(file_path):
+        os.remove(file_path)
+
+    domains = ["example.org"]
+    processes = ["Launch"]
+    emails = ["foobar@example.org"]
+    filenames = ["/var/foobar/txt"]
+
+    res = []
+    malware = Malware(name="TestMalware", is_family=False, description="")
+    res.append(malware)
+    for d in domains:
+        i = Indicator(indicator_types=["malicious-activity"], pattern="[domain-name:value='{}']".format(d), pattern_type="stix")
+        res.append(i)
+        res.append(Relationship(i, "indicates", malware))
+
+    for p in processes:
+        i = Indicator(indicator_types=["malicious-activity"], pattern="[process:name='{}']".format(p), pattern_type="stix")
+        res.append(i)
+        res.append(Relationship(i, "indicates", malware))
+
+    for f in filenames:
+        i = Indicator(indicator_types=["malicious-activity"], pattern="[file:name='{}']".format(f), pattern_type="stix")
+        res.append(i)
+        res.append(Relationship(i, "indicates", malware))
+
+    for e in emails:
+        i = Indicator(indicator_types=["malicious-activity"], pattern="[email-addr:value='{}']".format(e), pattern_type="stix")
+        res.append(i)
+        res.append(Relationship(i, "indicates", malware))
+
+    bundle = Bundle(objects=res)
+    with open(file_path, "w+") as f:
+        f.write(bundle.serialize(pretty=True))
+
+
+if __name__ == "__main__":
+    generate_test_stix_file("test.stix2")
+    print("test.stix2 file created")

tests/common/test_indicators.py

@@ -0,0 +1,32 @@
+# 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 logging
+import os
+
+from mvt.common.indicators import Indicators
+
+
+class TestIndicators:
+    def test_parse_stix2(self, indicator_file):
+        ind = Indicators(log=logging)
+        ind.load_indicators_files([indicator_file], load_default=False)
+        assert ind.ioc_files[0]["count"] == 4
+        assert len(ind.ioc_files[0]["domains"]) == 1
+        assert len(ind.ioc_files[0]["emails"]) == 1
+        assert len(ind.ioc_files[0]["file_names"]) == 1
+        assert len(ind.ioc_files[0]["processes"]) == 1
+
+    def test_check_domain(self, indicator_file):
+        ind = Indicators(log=logging)
+        ind.load_indicators_files([indicator_file], load_default=False)
+        assert ind.check_domain("https://www.example.org/foobar")
+        assert ind.check_domain("http://example.org:8080/toto")
+
+    def test_env_stix(self, indicator_file):
+        os.environ["MVT_STIX2"] = indicator_file
+        ind = Indicators(log=logging)
+        ind.load_indicators_files([], load_default=False)
+        assert ind.total_ioc_count == 4

tests/conftest.py

@@ -0,0 +1,26 @@
+# 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 os
+
+import pytest
+
+from .artifacts.generate_stix import generate_test_stix_file
+
+
+@pytest.fixture(scope="session", autouse=True)
+def indicator_file(request, tmp_path_factory):
+    indicator_dir = tmp_path_factory.mktemp("indicators")
+    stix_path = indicator_dir / "indicators.stix2"
+    generate_test_stix_file(stix_path)
+    return str(stix_path)
+
+
+@pytest.fixture(scope="session", autouse=True)
+def clean_test_env(request, tmp_path_factory):
+    try:
+        del os.environ["MVT_STIX2"]
+    except KeyError:
+        pass

tests/ios/test_backup_info.py

@@ -0,0 +1,19 @@
+# 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 logging
+
+from mvt.common.module import run_module
+from mvt.ios.modules.backup.backup_info import BackupInfo
+
+from ..utils import get_backup_folder
+
+
+class TestBackupInfoModule:
+    def test_manifest(self):
+        m = BackupInfo(base_folder=get_backup_folder(), log=logging)
+        run_module(m)
+        assert m.results["Build Version"] == "18C66"
+        assert m.results["IMEI"] == "42"

tests/ios/test_datausage.py

@@ -0,0 +1,31 @@
+# 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 logging
+
+from mvt.common.indicators import Indicators
+from mvt.common.module import run_module
+from mvt.ios.modules.mixed.net_datausage import Datausage
+
+from ..utils import get_backup_folder
+
+
+class TestDatausageModule:
+    def test_datausage(self):
+        m = Datausage(base_folder=get_backup_folder(), log=logging, results=[])
+        run_module(m)
+        assert len(m.results) == 42
+        assert len(m.timeline) == 60
+        assert len(m.detected) == 0
+
+    def test_detection(self, indicator_file):
+        m = Datausage(base_folder=get_backup_folder(), log=logging, results=[])
+        ind = Indicators(log=logging)
+        ind.parse_stix2(indicator_file)
+        # Adds a file that exists in the manifest.
+        ind.ioc_files[0]["processes"].append("CumulativeUsageTracker")
+        m.indicators = ind
+        run_module(m)
+        assert len(m.detected) == 2

tests/ios/test_manifest.py

@@ -0,0 +1,30 @@
+# 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 logging
+
+from mvt.common.indicators import Indicators
+from mvt.common.module import run_module
+from mvt.ios.modules.backup.manifest import Manifest
+
+from ..utils import get_backup_folder
+
+
+class TestManifestModule:
+    def test_manifest(self):
+        m = Manifest(base_folder=get_backup_folder(), log=logging, results=[])
+        run_module(m)
+        assert len(m.results) == 3721
+        assert len(m.timeline) == 5881
+        assert len(m.detected) == 0
+
+    def test_detection(self, indicator_file):
+        m = Manifest(base_folder=get_backup_folder(), log=logging, results=[])
+        ind = Indicators(log=logging)
+        ind.parse_stix2(indicator_file)
+        ind.ioc_files[0]["file_names"].append("com.apple.CoreBrightness.plist")
+        m.indicators = ind
+        run_module(m)
+        assert len(m.detected) == 1

tests/ios/test_tcc.py

@@ -0,0 +1,36 @@
+# 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 logging
+
+from mvt.common.indicators import Indicators
+from mvt.common.module import run_module
+from mvt.ios.modules.mixed.tcc import TCC
+
+from ..utils import get_backup_folder
+
+
+class TestTCCtModule:
+    def test_tcc(self):
+        m = TCC(base_folder=get_backup_folder(), log=logging, results=[])
+        run_module(m)
+        assert len(m.results) == 11
+        assert len(m.timeline) == 11
+        assert len(m.detected) == 0
+        assert m.results[0]["service"] == "kTCCServiceUbiquity"
+        assert m.results[0]["client"] == "com.apple.Preferences"
+        assert m.results[0]["auth_value"] == "allowed"
+
+    def test_tcc_detection(self, indicator_file):
+        m = TCC(base_folder=get_backup_folder(), log=logging, results=[])
+        ind = Indicators(log=logging)
+        ind.parse_stix2(indicator_file)
+        m.indicators = ind
+        run_module(m)
+        assert len(m.results) == 11
+        assert len(m.timeline) == 11
+        assert len(m.detected) == 1
+        assert m.detected[0]["service"] == "kTCCServiceLiverpool"
+        assert m.detected[0]["client"] == "Launch"

tests/utils.py

@@ -0,0 +1,28 @@
+# 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 os
+
+
+def get_artifact(fname):
+    """
+    Return the artifact path in the artifact folder
+    """
+    fpath = os.path.join(get_artifact_folder(), fname)
+    if os.path.isfile(fpath):
+        return fpath
+    return
+
+
+def get_artifact_folder():
+    return os.path.join(os.path.dirname(__file__), "artifacts")
+
+
+def get_backup_folder():
+    return os.path.join(os.path.dirname(__file__), "artifacts", "ios_backup")
+
+
+def get_indicator_file():
+    print("PYTEST env", os.getenv("PYTEST_CURRENT_TEST"))