Bumped version

Add initial implementation of SMS extraction using ADB

.github/workflows/python-package.yml

@@ -1,7 +1,7 @@
 # This workflow will install Python dependencies, run tests and lint with a variety of Python versions
 # For more information see: https://help.github.com/actions/language-and-framework-guides/using-python-with-github-actions
 
-name: Python package
+name: CI
 
 on:
   push:

README.md

@@ -6,6 +6,8 @@
 
 [![](https://img.shields.io/pypi/v/mvt)](https://pypi.org/project/mvt/)
 [![Documentation Status](https://readthedocs.org/projects/mvt/badge/?version=latest)](https://docs.mvt.re/en/latest/?badge=latest)
+[![CI](https://github.com/mvt-project/mvt/actions/workflows/python-package.yml/badge.svg)](https://github.com/mvt-project/mvt/actions/workflows/python-package.yml)
+[![Downloads](https://pepy.tech/badge/mvt)](https://pepy.tech/project/mvt)
 
 Mobile Verification Toolkit (MVT) is a collection of utilities to simplify and automate the process of gathering forensic traces helpful to identify a potential compromise of Android and iOS devices.
 

docs/android/backup.md

@@ -11,18 +11,37 @@ That said, most versions of Android should still allow to locally backup SMS mes
 Because `mvt-android check-backup` currently only supports checking SMS messages, you can indicate to backup only those:
 
 ```bash
-adb backup com.android.providers.telephony
+adb backup -nocompress com.android.providers.telephony
 ```
 
 In case you nonetheless wish to take a full backup, you can do so with
 
 ```bash
-adb backup -all
+adb backup -nocompress -all
 ```
 
-## Unpack the backup
+Some recent phones will enforce the utilisation of a password to encrypt the backup archive. In that case, the password will obviously be needed to extract and analyse the data later on.
 
-In order to unpack the backup, use [Android Backup Extractor (ABE)](https://github.com/nelenkov/android-backup-extractor) to convert it to a readable file format. Make sure that java is installed on your system and use the following command:
+## Unpack and check the backup
+
+MVT includes a partial implementation of the Android Backup parsing, because of the implementation difference in the compression algorithm between Java and Python. The `-nocompress` option passed to adb in the section above allows to avoid this issue. You can analyse and extract SMSs containing links from the backup directly with MVT:
+
+```bash
+$ mvt-android check-backup --output /path/to/results/ /path/to/backup.ab
+14:09:45 INFO     [mvt.android.cli] Checking ADB backup located at: backup.ab
+         INFO     [mvt.android.modules.backup.sms] Running module SMS...
+         INFO     [mvt.android.modules.backup.sms] Processing SMS backup file at
+                  apps/com.android.providers.telephony/d_f/000000_sms_backup
+         INFO     [mvt.android.modules.backup.sms] Extracted a total of 64 SMS messages containing links
+```
+
+If the backup is encrypted, MVT will prompt you to enter the password.
+
+Through the `--iocs` argument you can specify a [STIX2](https://oasis-open.github.io/cti-documentation/stix/intro) file defining a list of malicious indicators to check against the records extracted from the backup by MVT. Any matches will be highlighted in the terminal output.
+
+## Alternative ways to unpack and check the backup
+
+If you encounter an issue during the analysis of the backup, you can alternatively use [Android Backup Extractor (ABE)](https://github.com/nelenkov/android-backup-extractor) to convert it to a readable file format. Make sure that java is installed on your system and use the following command:
 
 ```bash
 java -jar ~/path/to/abe.jar unpack backup.ab backup.tar
@@ -33,17 +52,4 @@ If the backup is encrypted, ABE will prompt you to enter the password.
 
 Alternatively, [ab-decrypt](https://github.com/joernheissler/ab-decrypt) can be used for that purpose.
 
-## Check the backup
-
-You can then extract SMSs containing links with MVT:
-
-```bash
-$ mvt-android check-backup --output /path/to/results/ /path/to/backup/
-16:18:38 INFO     [mvt.android.cli] Checking ADB backup located at: .
-         INFO     [mvt.android.modules.backup.sms] Running module SMS...
-         INFO     [mvt.android.modules.backup.sms] Processing SMS backup file at /path/to/backup/apps/com.android.providers.telephony/d_f/000000_sms_backup
-16:18:39 INFO     [mvt.android.modules.backup.sms] Extracted a total of
-                  64 SMS messages containing links
-```
-
-Through the `--iocs` argument you can specify a [STIX2](https://oasis-open.github.io/cti-documentation/stix/intro) file defining a list of malicious indicators to check against the records extracted from the backup by MVT. Any matches will be highlighted in the terminal output.
+You can then extract SMSs containing links with MVT by passing the folder path as parameter instead of the `.ab` file: `mvt-android check-backup --output /path/to/results/ /path/to/backup/`.

docs/ios/records.md

@@ -190,7 +190,7 @@ If indicators are provided through the command-line, they are checked against th
     Backup: :material-check:
     Full filesystem dump: :material-check:
 
-This JSON file is created by mvt-ios' `Datausage` module. The module extracts records from a SQLite database located */private/var/wireless/Library/Databases/DataUsage.sqlite*, which contains a history of data usage by processes running on the system. Besides the network statistics, these records are particularly important because they might show traces of malicious process executions and the relevant timeframe. In particular, processes which do not have a valid bundle ID might require particular attention.
+This JSON file is created by mvt-ios' `Datausage` module. The module extracts records from a SQLite database located */private/var/wireless/Library/Databases/DataUsage.sqlite*, which contains a history of network data usage by processes running on the system. It does not log network traffic through WiFi (the fields `WIFI_IN` and `WIFI_OUT` are always empty), and the `WWAN_IN` and `WWAN_OUT` fields are stored in bytes. Besides the network statistics, these records are particularly important because they might show traces of malicious process executions and the relevant timeframe. In particular, processes which do not have a valid bundle ID might require particular attention.
 
 If indicators are provided through the command-line, they are checked against the process names. Any matches are stored in *datausage_detected.json*. If running on a full filesystem dump and if the `--fast` flag was not enabled by command-line, mvt-ios will highlight processes which look suspicious and check the presence of a binary file of the same name in the dump.
 

mvt/android/cli.py

@@ -3,12 +3,20 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
+import getpass
+import io
 import logging
 import os
+import tarfile
+from pathlib import Path
+from zipfile import ZipFile
 
 import click
 from rich.logging import RichHandler
 
+from mvt.android.parsers.backup import (AndroidBackupParsingError,
+                                        InvalidBackupPassword, parse_ab_header,
+                                        parse_backup_file)
 from mvt.common.help import (HELP_MSG_FAST, HELP_MSG_IOC,
                              HELP_MSG_LIST_MODULES, HELP_MSG_MODULE,
                              HELP_MSG_OUTPUT, HELP_MSG_SERIAL)
@@ -21,6 +29,7 @@ from .lookups.koodous import koodous_lookup
 from .lookups.virustotal import virustotal_lookup
 from .modules.adb import ADB_MODULES
 from .modules.backup import BACKUP_MODULES
+from .modules.bugreport import BUGREPORT_MODULES
 
 # Setup logging using Rich.
 LOG_FORMAT = "[%(name)s] %(message)s"
@@ -46,9 +55,9 @@ def version():
 
 
 #==============================================================================
-# Download APKs
+# Command: download-apks
 #==============================================================================
-@cli.command("download-apks", help="Download all or non-safelisted installed APKs installed on the device")
+@cli.command("download-apks", help="Download all or only non-system installed APKs")
 @click.option("--serial", "-s", type=str, help=HELP_MSG_SERIAL)
 @click.option("--all-apks", "-a", is_flag=True,
               help="Extract all packages installed on the phone, including system packages")
@@ -99,7 +108,7 @@ def download_apks(ctx, all_apks, virustotal, koodous, all_checks, output, from_f
 
 
 #==============================================================================
-# Checks through ADB
+# Command: check-adb
 #==============================================================================
 @cli.command("check-adb", help="Check an Android device over adb")
 @click.option("--serial", "-s", type=str, help=HELP_MSG_SERIAL)
@@ -157,17 +166,25 @@ def check_adb(ctx, iocs, output, fast, list_modules, module, serial):
 
 
 #==============================================================================
-# Check ADB backup
+# Command: check-bugreport
 #==============================================================================
-@cli.command("check-backup", help="Check an Android Backup")
-@click.option("--serial", "-s", type=str, help=HELP_MSG_SERIAL)
+@cli.command("check-bugreport", help="Check an Android Bug Report")
 @click.option("--iocs", "-i", type=click.Path(exists=True), multiple=True,
               default=[], help=HELP_MSG_IOC)
 @click.option("--output", "-o", type=click.Path(exists=False), help=HELP_MSG_OUTPUT)
-@click.argument("BACKUP_PATH", type=click.Path(exists=True))
+@click.option("--list-modules", "-l", is_flag=True, help=HELP_MSG_LIST_MODULES)
+@click.option("--module", "-m", help=HELP_MSG_MODULE)
+@click.argument("BUGREPORT_PATH", type=click.Path(exists=True))
 @click.pass_context
-def check_backup(ctx, iocs, output, backup_path, serial):
-    log.info("Checking ADB backup located at: %s", backup_path)
+def check_bugreport(ctx, iocs, output, list_modules, module, bugreport_path):
+    if list_modules:
+        log.info("Following is the list of available check-bugreport modules:")
+        for adb_module in BUGREPORT_MODULES:
+            log.info(" - %s", adb_module.__name__)
+
+        return
+
+    log.info("Checking an Android Bug Report located at: %s", bugreport_path)
 
     if output and not os.path.exists(output):
         try:
@@ -179,14 +196,110 @@ def check_backup(ctx, iocs, output, backup_path, serial):
     indicators = Indicators(log=log)
     indicators.load_indicators_files(iocs)
 
-    if os.path.isfile(backup_path):
-        log.critical("The path you specified is a not a folder!")
+    if os.path.isfile(bugreport_path):
+        bugreport_format = "zip"
+        zip_archive = ZipFile(bugreport_path)
+        zip_files = []
+        for file_name in zip_archive.namelist():
+            zip_files.append(file_name)
+    elif os.path.isdir(bugreport_path):
+        bugreport_format = "dir"
+        folder_files = []
+        parent_path = Path(bugreport_path).absolute().as_posix()
+        for root, subdirs, subfiles in os.walk(os.path.abspath(bugreport_path)):
+            for file_name in subfiles:
+                folder_files.append(os.path.relpath(os.path.join(root, file_name), parent_path))
 
-        if os.path.basename(backup_path) == "backup.ab":
-            log.info("You can use ABE (https://github.com/nelenkov/android-backup-extractor) "
-                     "to extract 'backup.ab' files!")
+    timeline = []
+    timeline_detected = []
+    for bugreport_module in BUGREPORT_MODULES:
+        if module and bugreport_module.__name__ != module:
+            continue
+
+        m = bugreport_module(base_folder=bugreport_path, output_folder=output,
+                             log=logging.getLogger(bugreport_module.__module__))
+
+        if bugreport_format == "zip":
+            m.from_zip(zip_archive, zip_files)
+        else:
+            m.from_folder(bugreport_path, folder_files)
+
+        if indicators.total_ioc_count:
+            m.indicators = indicators
+            m.indicators.log = m.log
+
+        run_module(m)
+        timeline.extend(m.timeline)
+        timeline_detected.extend(m.timeline_detected)
+
+    if output:
+        if len(timeline) > 0:
+            save_timeline(timeline, os.path.join(output, "timeline.csv"))
+        if len(timeline_detected) > 0:
+            save_timeline(timeline_detected, os.path.join(output, "timeline_detected.csv"))
+
+
+#==============================================================================
+# Command: check-backup
+#==============================================================================
+@cli.command("check-backup", help="Check an Android Backup")
+@click.option("--serial", "-s", type=str, help=HELP_MSG_SERIAL)
+@click.option("--iocs", "-i", type=click.Path(exists=True), multiple=True,
+              default=[], help=HELP_MSG_IOC)
+@click.option("--output", "-o", type=click.Path(exists=False), help=HELP_MSG_OUTPUT)
+@click.argument("BACKUP_PATH", type=click.Path(exists=True))
+@click.pass_context
+def check_backup(ctx, iocs, output, backup_path, serial):
+    log.info("Checking ADB backup located at: %s", backup_path)
+
+    if os.path.isfile(backup_path):
+        # AB File
+        backup_type = "ab"
+        with open(backup_path, "rb") as handle:
+            data = handle.read()
+        header = parse_ab_header(data)
+        if not header["backup"]:
+            log.critical("Invalid backup format, file should be in .ab format")
+            ctx.exit(1)
+        password = None
+        if header["encryption"] != "none":
+            password = getpass.getpass(prompt="Backup Password: ", stream=None)
+        try:
+            tardata = parse_backup_file(data, password=password)
+        except InvalidBackupPassword:
+            log.critical("Invalid backup password")
+            ctx.exit(1)
+        except AndroidBackupParsingError:
+            log.critical("Impossible to parse this backup file, please use Android Backup Extractor instead")
+            ctx.exit(1)
+
+        dbytes = io.BytesIO(tardata)
+        tar = tarfile.open(fileobj=dbytes)
+        files = []
+        for member in tar:
+            files.append(member.name)
+
+    elif os.path.isdir(backup_path):
+        backup_type = "folder"
+        backup_path = Path(backup_path).absolute().as_posix()
+        files = []
+        for root, subdirs, subfiles in os.walk(os.path.abspath(backup_path)):
+            for fname in subfiles:
+                files.append(os.path.relpath(os.path.join(root, fname), backup_path))
+    else:
+        log.critical("Invalid backup path, path should be a folder or an Android Backup (.ab) file")
         ctx.exit(1)
 
+    if output and not os.path.exists(output):
+        try:
+            os.makedirs(output)
+        except Exception as e:
+            log.critical("Unable to create output folder %s: %s", output, e)
+            ctx.exit(1)
+
+    indicators = Indicators(log=log)
+    indicators.load_indicators_files(iocs)
+
     for module in BACKUP_MODULES:
         m = module(base_folder=backup_path, output_folder=output,
                    log=logging.getLogger(module.__module__))
@@ -196,6 +309,11 @@ def check_backup(ctx, iocs, output, backup_path, serial):
         if serial:
             m.serial = serial
 
+        if backup_type == "folder":
+            m.from_folder(backup_path, files)
+        else:
+            m.from_ab(backup_path, tar, files)
+
         run_module(m)
 
 

mvt/android/modules/adb/__init__.py

@@ -17,11 +17,13 @@ from .logcat import Logcat
 from .packages import Packages
 from .processes import Processes
 from .root_binaries import RootBinaries
+from .selinux_status import SELinuxStatus
 from .settings import Settings
 from .sms import SMS
 from .whatsapp import Whatsapp
 
 ADB_MODULES = [ChromeHistory, SMS, Whatsapp, Processes, Getprop, Settings,
-               DumpsysBatteryHistory, DumpsysBatteryDaily, DumpsysReceivers,
-               DumpsysActivities, DumpsysAccessibility, DumpsysDBInfo,
-               DumpsysFull, Packages, RootBinaries, Logcat, Files]
+               SELinuxStatus, DumpsysBatteryHistory, DumpsysBatteryDaily,
+               DumpsysReceivers, DumpsysActivities, DumpsysAccessibility,
+               DumpsysDBInfo, DumpsysFull, Packages, Logcat, RootBinaries,
+               Files]

mvt/android/modules/adb/base.py

@@ -3,6 +3,8 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
+import base64
+import getpass
 import logging
 import os
 import random
@@ -18,6 +20,8 @@ from adb_shell.exceptions import (AdbCommandFailureException, DeviceAuthError,
                                   UsbDeviceNotFoundError, UsbReadFailedError)
 from usb1 import USBErrorAccess, USBErrorBusy
 
+from mvt.android.parsers.backup import (InvalidBackupPassword, parse_ab_header,
+                                        parse_backup_file)
 from mvt.common.module import InsufficientPrivileges, MVTModule
 
 log = logging.getLogger(__name__)
@@ -243,6 +247,34 @@ class AndroidExtraction(MVTModule):
         # Disconnect from the device.
         self._adb_disconnect()
 
+    def _generate_backup(self, package_name):
+        # Run ADB command to create a backup of SMS app
+        self.log.warning("Please check phone and accept Android backup prompt. You may need to set a backup password. \a")
+
+        # Run ADB command to create a backup of SMS app
+        # TODO: Base64 encoding as temporary fix to avoid byte-mangling over the shell transport...
+        backup_output_b64 = self._adb_command("/system/bin/bu backup -nocompress '{}' | base64".format(package_name))
+        backup_output = base64.b64decode(backup_output_b64)
+        header = parse_ab_header(backup_output)
+
+        if not header["backup"]:
+            self.log.error("Extracting SMS via Android backup failed. No valid backup data found.")
+            return
+
+        if header["encryption"] == "none":
+            return parse_backup_file(backup_output, password=None)
+
+        # Backup encrypted. Request password from user.
+        for password_retry in range(0, 3):
+            backup_password = getpass.getpass(prompt="Backup Password: ", stream=None)
+            try:
+                decrypted_backup_tar = parse_backup_file(backup_output, backup_password)
+                return decrypted_backup_tar
+            except InvalidBackupPassword:
+                self.log.info("Invalid backup password")
+
+        self.log.warn("All attempts to decrypt backup with password failed!")
+
     def run(self):
         """Run the main procedure."""
         raise NotImplementedError

mvt/android/modules/adb/chrome_history.py

@@ -68,7 +68,7 @@ class ChromeHistory(AndroidExtraction):
                 "url": item[1],
                 "visit_id": item[2],
                 "timestamp": item[3],
-                "isodate": convert_timestamp_to_iso(convert_chrometime_to_unix[item[3]]),
+                "isodate": convert_timestamp_to_iso(convert_chrometime_to_unix(item[3])),
                 "redirect_source": item[4],
             })
 
@@ -78,5 +78,8 @@ class ChromeHistory(AndroidExtraction):
         log.info("Extracted a total of %d history items", len(self.results))
 
     def run(self):
-        self._adb_process_file(os.path.join("/", CHROME_HISTORY_PATH),
-                               self._parse_db)
+        try:
+            self._adb_process_file(os.path.join("/", CHROME_HISTORY_PATH),
+                                   self._parse_db)
+        except Exception as e:
+            self.log.error(e)

mvt/android/modules/adb/dumpsys_accessibility.py

@@ -5,6 +5,8 @@
 
 import logging
 
+from mvt.android.parsers import parse_dumpsys_accessibility
+
 from .base import AndroidExtraction
 
 log = logging.getLogger(__name__)
@@ -20,45 +22,24 @@ class DumpsysAccessibility(AndroidExtraction):
                          log=log, results=results)
 
     def check_indicators(self):
+        if not self.indicators:
+            return
+
         for result in self.results:
-            ioc = self.indicators.check_app_id(result["package"])
+            ioc = self.indicators.check_app_id(result["package_name"])
             if ioc:
                 result["matched_indicator"] = ioc
                 self.detected.append(result)
                 continue
 
-    @staticmethod
-    def parse_accessibility(output):
-        results = []
-
-        in_services = False
-        for line in output.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)
-
-            results.append({
-                "package": service.split("/")[0],
-                "service": service,
-            })
-
-        return results
-
     def run(self):
         self._adb_connect()
-
         output = self._adb_command("dumpsys accessibility")
-        self.results = self.parse_accessibility(output)
+        self._adb_disconnect()
+
+        self.results = parse_dumpsys_accessibility(output)
+
+        for result in self.results:
+            log.info("Found installed accessibility service \"%s\"", result.get("service"))
 
         self.log.info("Identified a total of %d accessibility services", len(self.results))
-
-        self._adb_disconnect()

mvt/android/modules/adb/dumpsys_activities.py

@@ -5,6 +5,8 @@
 
 import logging
 
+from mvt.android.parsers import parse_dumpsys_activity_resolver_table
+
 from .base import AndroidExtraction
 
 log = logging.getLogger(__name__)
@@ -22,74 +24,22 @@ class DumpsysActivities(AndroidExtraction):
         self.results = results if results else {}
 
     def check_indicators(self):
+        if not self.indicators:
+            return
+
         for intent, activities in self.results.items():
             for activity in activities:
-                ioc = self.indicators.check_app_id(activity["package"])
+                ioc = self.indicators.check_app_id(activity["package_name"])
                 if ioc:
                     activity["matched_indicator"] = ioc
                     self.detected.append({intent: activity})
                     continue
 
-    @staticmethod
-    def parse_activity_resolver_table(output):
-        results = {}
-
-        in_activity_resolver_table = False
-        in_non_data_actions = False
-        intent = None
-        for line in output.split("\n"):
-            if line.startswith("Activity Resolver Table:"):
-                in_activity_resolver_table = True
-                continue
-
-            if not in_activity_resolver_table:
-                continue
-
-            if line.startswith("  Non-Data Actions:"):
-                in_non_data_actions = True
-                continue
-
-            if not in_non_data_actions:
-                continue
-
-            # If we hit an empty line, the Non-Data Actions section should be
-            # finished.
-            if line.strip() == "":
-                break
-
-            # We detect the action name.
-            if line.startswith(" " * 6) and not line.startswith(" " * 8) and ":" in line:
-                intent = line.strip().replace(":", "")
-                results[intent] = []
-                continue
-
-            # If we are not in an intent block yet, skip.
-            if not intent:
-                continue
-
-            # If we are in a block but the line does not start with 8 spaces
-            # it means the block ended a new one started, so we reset and
-            # continue.
-            if not line.startswith(" " * 8):
-                intent = None
-                continue
-
-            # If we got this far, we are processing receivers for the
-            # activities we are interested in.
-            activity = line.strip().split(" ")[1]
-            package = activity.split("/")[0]
-
-            results[intent].append({
-                "package": package,
-                "activity": activity,
-            })
-
-        return results
-
     def run(self):
         self._adb_connect()
-
         output = self._adb_command("dumpsys package")
-        self.results = self.parse_activity_resolver_table(output)
-
         self._adb_disconnect()
+
+        self.results = parse_dumpsys_activity_resolver_table(output)
+
+        self.log.info("Extracted activities for %d intents", len(self.results))

mvt/android/modules/adb/dumpsys_battery_daily.py

@@ -5,6 +5,8 @@
 
 import logging
 
+from mvt.android.parsers import parse_dumpsys_battery_daily
+
 from .base import AndroidExtraction
 
 log = logging.getLogger(__name__)
@@ -24,67 +26,25 @@ class DumpsysBatteryDaily(AndroidExtraction):
             "timestamp": record["from"],
             "module": self.__class__.__name__,
             "event": "battery_daily",
-            "data": f"Recorded update of package {record['package']} with vers {record['vers']}"
+            "data": f"Recorded update of package {record['package_name']} with vers {record['vers']}"
         }
 
     def check_indicators(self):
+        if not self.indicators:
+            return
+
         for result in self.results:
-            ioc = self.indicators.check_app_id(result["package"])
+            ioc = self.indicators.check_app_id(result["package_name"])
             if ioc:
                 result["matched_indicator"] = ioc
                 self.detected.append(result)
                 continue
 
-    @staticmethod
-    def parse_battery_history(output):
-        results = []
-        daily = None
-        daily_updates = []
-        for line in output.split("\n")[1:]:
-            if line.startswith("  Daily from "):
-                timeframe = line[13:].strip()
-                date_from, date_to = timeframe.strip(":").split(" to ", 1)
-                daily = {"from": date_from[0:10], "to": date_to[0:10]}
-
-            if not daily:
-                continue
-
-            if line.strip() == "":
-                results.extend(daily_updates)
-                daily = None
-                daily_updates = []
-                continue
-
-            if not line.strip().startswith("Update "):
-                continue
-
-            line = line.strip().replace("Update ", "")
-            package, vers = line.split(" ", 1)
-            vers_nr = vers.split("=", 1)[1]
-
-            already_seen = False
-            for update in daily_updates:
-                if package == update["package"] and vers_nr == update["vers"]:
-                    already_seen = True
-                    break
-
-            if not already_seen:
-                daily_updates.append({
-                    "action": "update",
-                    "from": daily["from"],
-                    "to": daily["to"],
-                    "package": package,
-                    "vers": vers_nr,
-                })
-
-        return results
-
     def run(self):
         self._adb_connect()
-
         output = self._adb_command("dumpsys batterystats --daily")
-        self.results = self.parse_battery_history(output)
+        self._adb_disconnect()
+
+        self.results = parse_dumpsys_battery_daily(output)
 
         self.log.info("Extracted %d records from battery daily stats", len(self.results))
-
-        self._adb_disconnect()

mvt/android/modules/adb/dumpsys_battery_history.py

@@ -5,6 +5,8 @@
 
 import logging
 
+from mvt.android.parsers import parse_dumpsys_battery_history
+
 from .base import AndroidExtraction
 
 log = logging.getLogger(__name__)
@@ -20,69 +22,21 @@ class DumpsysBatteryHistory(AndroidExtraction):
                          log=log, results=results)
 
     def check_indicators(self):
+        if not self.indicators:
+            return
+
         for result in self.results:
-            ioc = self.indicators.check_app_id(result["package"])
+            ioc = self.indicators.check_app_id(result["package_name"])
             if ioc:
                 result["matched_indicator"] = ioc
                 self.detected.append(result)
                 continue
 
-    @staticmethod
-    def parse_battery_history(output):
-        results = []
-
-        for line in output.split("\n")[1:]:
-            if line.strip() == "":
-                break
-
-            time_elapsed, rest = line.strip().split(" ", 1)
-
-            start = line.find(" 100 ")
-            if start == -1:
-                continue
-
-            line = line[start+5:]
-
-            event = ""
-            if line.startswith("+job"):
-                event = "start_job"
-            elif line.startswith("-job"):
-                event = "end_job"
-            elif line.startswith("+running +wake_lock="):
-                event = "wake"
-            else:
-                continue
-
-            if event in ["start_job", "end_job"]:
-                uid = line[line.find("=")+1:line.find(":")]
-                service = line[line.find(":")+1:].strip('"')
-                package = service.split("/")[0]
-            elif event == "wake":
-                uid = line[line.find("=")+1:line.find(":")]
-                service = line[line.find("*walarm*:")+9:].split(" ")[0].strip('"').strip()
-                if service == "" or "/" not in service:
-                    continue
-
-                package = service.split("/")[0]
-            else:
-                continue
-
-            results.append({
-                "time_elapsed": time_elapsed,
-                "event": event,
-                "uid": uid,
-                "package": package,
-                "service": service,
-            })
-
-        return results
-
     def run(self):
         self._adb_connect()
-
         output = self._adb_command("dumpsys batterystats --history")
-        self.results = self.parse_battery_history(output)
+        self._adb_disconnect()
+
+        self.results = parse_dumpsys_battery_history(output)
 
         self.log.info("Extracted %d records from battery history", len(self.results))
-
-        self._adb_disconnect()

mvt/android/modules/adb/dumpsys_dbinfo.py

@@ -6,6 +6,8 @@
 import logging
 import re
 
+from mvt.android.parsers import parse_dumpsys_dbinfo
+
 from .base import AndroidExtraction
 
 log = logging.getLogger(__name__)
@@ -23,6 +25,9 @@ class DumpsysDBInfo(AndroidExtraction):
                          log=log, results=results)
 
     def check_indicators(self):
+        if not self.indicators:
+            return
+
         for result in self.results:
             path = result.get("path", "")
             for part in path.split("/"):
@@ -32,47 +37,12 @@ class DumpsysDBInfo(AndroidExtraction):
                     self.detected.append(result)
                     continue
 
-    @staticmethod
-    def parse_dbinfo(output):
-        results = []
-
-        rxp = re.compile(r'.*\[([0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3})\].*\[Pid:\((\d+)\)\](\w+).*sql\=\"(.+?)\".*path\=(.*?$)')
-
-        in_operations = False
-        for line in output.split("\n"):
-            if line.strip() == "Most recently executed operations:":
-                in_operations = True
-                continue
-
-            if not in_operations:
-                continue
-
-            if not line.startswith("        "):
-                in_operations = False
-                continue
-
-            matches = rxp.findall(line)
-            if not matches:
-                continue
-
-            match = matches[0]
-            results.append({
-                "isodate": match[0],
-                "pid": match[1],
-                "action": match[2],
-                "sql": match[3],
-                "path": match[4],
-            })
-
-        return results
-
     def run(self):
         self._adb_connect()
-
         output = self._adb_command("dumpsys dbinfo")
-        self.results = self.parse_dbinfo(output)
+        self._adb_disconnect()
+
+        self.results = parse_dumpsys_dbinfo(output)
 
         self.log.info("Extracted a total of %d records from database information",
                       len(self.results))
-
-        self._adb_disconnect()

mvt/android/modules/adb/dumpsys_receivers.py

@@ -5,6 +5,8 @@
 
 import logging
 
+from mvt.android.parsers import parse_dumpsys_receiver_resolver_table
+
 from .base import AndroidExtraction
 
 log = logging.getLogger(__name__)
@@ -28,6 +30,9 @@ class DumpsysReceivers(AndroidExtraction):
         self.results = results if results else {}
 
     def check_indicators(self):
+        if not self.indicators:
+            return
+
         for intent, receivers in self.results.items():
             for receiver in receivers:
                 if intent == INTENT_NEW_OUTGOING_SMS:
@@ -46,72 +51,16 @@ class DumpsysReceivers(AndroidExtraction):
                     self.log.info("Found a receiver monitoring outgoing calls: \"%s\"",
                                   receiver["receiver"])
 
-            ioc = self.indicators.check_app_id(receiver["package"])
+            ioc = self.indicators.check_app_id(receiver["package_name"])
             if ioc:
                 receiver["matched_indicator"] = ioc
                 self.detected.append({intent: receiver})
                 continue
 
-    @staticmethod
-    def parse_receiver_resolver_table(output):
-        results = {}
-
-        in_receiver_resolver_table = False
-        in_non_data_actions = False
-        intent = None
-        for line in output.split("\n"):
-            if line.startswith("Receiver Resolver Table:"):
-                in_receiver_resolver_table = True
-                continue
-
-            if not in_receiver_resolver_table:
-                continue
-
-            if line.startswith("  Non-Data Actions:"):
-                in_non_data_actions = True
-                continue
-
-            if not in_non_data_actions:
-                continue
-
-            # If we hit an empty line, the Non-Data Actions section should be
-            # finished.
-            if line.strip() == "":
-                break
-
-            # We detect the action name.
-            if line.startswith(" " * 6) and not line.startswith(" " * 8) and ":" in line:
-                intent = line.strip().replace(":", "")
-                results[intent] = []
-                continue
-
-            # If we are not in an intent block yet, skip.
-            if not intent:
-                continue
-
-            # If we are in a block but the line does not start with 8 spaces
-            # it means the block ended a new one started, so we reset and
-            # continue.
-            if not line.startswith(" " * 8):
-                intent = None
-                continue
-
-            # If we got this far, we are processing receivers for the
-            # activities we are interested in.
-            receiver = line.strip().split(" ")[1]
-            package = receiver.split("/")[0]
-
-            results[intent].append({
-                "package": package,
-                "receiver": receiver,
-            })
-
-        return results
-
     def run(self):
         self._adb_connect()
 
         output = self._adb_command("dumpsys package")
-        self.results = self.parse_receiver_resolver_table(output)
+        self.results = parse_dumpsys_receiver_resolver_table(output)
 
         self._adb_disconnect()

mvt/android/modules/adb/files.py

@@ -24,9 +24,9 @@ class Files(AndroidExtraction):
                          log=log, results=results)
         self.full_find = False
 
-    def find_files(self, file_path):
+    def find_files(self, folder):
         if self.full_find:
-            output = self._adb_command(f"find '{file_path}' -printf '%T@ %m %s %u %g %p\n' 2> /dev/null")
+            output = self._adb_command(f"find '{folder}' -printf '%T@ %m %s %u %g %p\n' 2> /dev/null")
 
             for file_line in output.splitlines():
                 [unix_timestamp, mode, size, owner, group, full_path] = file_line.rstrip().split(" ", 5)
@@ -42,7 +42,7 @@ class Files(AndroidExtraction):
                     "group": group,
                 })
         else:
-            output = self._adb_command(f"find '{file_path}' 2> /dev/null")
+            output = self._adb_command(f"find '{folder}' 2> /dev/null")
             for file_line in output.splitlines():
                 self.results.append({"path": file_line.rstrip()})
 

mvt/android/modules/adb/getprop.py

@@ -5,6 +5,9 @@
 
 import logging
 import re
+from datetime import datetime, timedelta
+
+from mvt.android.parsers import parse_getprop
 
 from .base import AndroidExtraction
 
@@ -24,22 +27,17 @@ class Getprop(AndroidExtraction):
 
     def run(self):
         self._adb_connect()
-
-        rxp = re.compile(r"\[(.+?)\]: \[(.+?)\]")
-        out = self._adb_command("getprop")
-        for line in out.splitlines():
-            line = line.strip()
-            if line == "":
-                continue
-
-            matches = re.findall(rxp, line)
-            if not matches or len(matches[0]) != 2:
-                continue
-
-            key = matches[0][0]
-            value = matches[0][1]
-            self.results[key] = value
-
+        output = self._adb_command("getprop")
         self._adb_disconnect()
 
+        self.results = parse_getprop(output)
+
+        # Alert if phone is outdated.
+        security_patch = self.results.get("ro.build.version.security_patch", "")
+        if security_patch:
+            patch_date = datetime.strptime(security_patch, "%Y-%m-%d")
+            if (datetime.now() - patch_date) > timedelta(days=6*30):
+                self.log.warning("This phone has not received security updates for more than "
+                                 "six months (last update: %s)", security_patch)
+
         self.log.info("Extracted %d Android system properties", len(self.results))

mvt/android/modules/adb/packages.py

@@ -15,6 +15,30 @@ from .base import AndroidExtraction
 
 log = logging.getLogger(__name__)
 
+DANGEROUS_PERMISSIONS_THRESHOLD = 10
+DANGEROUS_PERMISSIONS = [
+    "android.permission.ACCESS_COARSE_LOCATION",
+    "android.permission.ACCESS_FINE_LOCATION",
+    "android.permission.AUTHENTICATE_ACCOUNTS",
+    "android.permission.CAMERA",
+    "android.permission.DISABLE_KEYGUARD",
+    "android.permission.PROCESS_OUTGOING_CALLS",
+    "android.permission.READ_CALENDAR",
+    "android.permission.READ_CALL_LOG",
+    "android.permission.READ_CONTACTS",
+    "android.permission.READ_PHONE_STATE",
+    "android.permission.READ_SMS",
+    "android.permission.RECEIVE_MMS",
+    "android.permission.RECEIVE_SMS",
+    "android.permission.RECEIVE_WAP_PUSH",
+    "android.permission.RECORD_AUDIO",
+    "android.permission.SEND_SMS",
+    "android.permission.SYSTEM_ALERT_WINDOW",
+    "android.permission.USE_CREDENTIALS",
+    "android.permission.USE_SIP",
+    "com.android.browser.permission.READ_HISTORY_BOOKMARKS",
+]
+
 
 class Packages(AndroidExtraction):
     """This module extracts the list of installed packages."""
@@ -47,7 +71,7 @@ class Packages(AndroidExtraction):
     def check_indicators(self):
         root_packages_path = os.path.join("..", "..", "data", "root_packages.txt")
         root_packages_string = pkg_resources.resource_string(__name__, root_packages_path)
-        root_packages = root_packages_string.decode("utf-8").split("\n")
+        root_packages = root_packages_string.decode("utf-8").splitlines()
         root_packages = [rp.strip() for rp in root_packages]
 
         for result in self.results:
@@ -57,6 +81,9 @@ class Packages(AndroidExtraction):
                 self.detected.append(result)
                 continue
 
+            if not self.indicators:
+                continue
+
             ioc = self.indicators.check_app_id(result.get("package_name"))
             if ioc:
                 result["matched_indicator"] = ioc
@@ -69,6 +96,46 @@ class Packages(AndroidExtraction):
                     result["matched_indicator"] = ioc
                     self.detected.append(result)
 
+    @staticmethod
+    def parse_package_for_details(output):
+        details = {
+            "uid": "",
+            "version_name": "",
+            "version_code": "",
+            "timestamp": "",
+            "first_install_time": "",
+            "last_update_time": "",
+            "requested_permissions": [],
+        }
+
+        in_permissions = False
+        for line in output.splitlines():
+            if in_permissions:
+                if line.startswith(" " * 4) and not line.startswith(" " * 6):
+                    in_permissions = False
+                    continue
+
+                permission = line.strip().split(":")[0]
+                details["requested_permissions"].append(permission)
+
+            if line.strip().startswith("userId="):
+                details["uid"] = line.split("=")[1].strip()
+            elif line.strip().startswith("versionName="):
+                details["version_name"] = line.split("=")[1].strip()
+            elif line.strip().startswith("versionCode="):
+                details["version_code"] = line.split("=", 1)[1].strip()
+            elif line.strip().startswith("timeStamp="):
+                details["timestamp"] = line.split("=")[1].strip()
+            elif line.strip().startswith("firstInstallTime="):
+                details["first_install_time"] = line.split("=")[1].strip()
+            elif line.strip().startswith("lastUpdateTime="):
+                details["last_update_time"] = line.split("=")[1].strip()
+            elif line.strip() == "requested permissions:":
+                in_permissions = True
+                continue
+
+        return details
+
     def _get_files_for_package(self, package_name):
         output = self._adb_command(f"pm path {package_name}")
         output = output.strip().replace("package:", "")
@@ -76,7 +143,7 @@ class Packages(AndroidExtraction):
             return []
 
         package_files = []
-        for file_path in output.split("\n"):
+        for file_path in output.splitlines():
             file_path = file_path.strip()
 
             md5 = self._adb_command(f"md5sum {file_path}").split(" ")[0]
@@ -97,11 +164,9 @@ class Packages(AndroidExtraction):
     def run(self):
         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")
+        packages = self._adb_command("pm list packages -u -i -f")
 
-        for line in packages.split("\n"):
+        for line in packages.splitlines():
             line = line.strip()
             if not line.startswith("package:"):
                 continue
@@ -117,31 +182,22 @@ class Packages(AndroidExtraction):
                 if installer == "null":
                     installer = None
 
-            try:
-                uid = fields[2].split(":")[1].strip()
-            except IndexError:
-                uid = None
-
-            dumpsys = self._adb_command(f"dumpsys package {package_name} | grep -A2 timeStamp").split("\n")
-            timestamp = dumpsys[0].split("=")[1].strip()
-            first_install = dumpsys[1].split("=")[1].strip()
-            last_update = dumpsys[2].split("=")[1].strip()
-
             package_files = self._get_files_for_package(package_name)
-
-            self.results.append({
+            new_package = {
                 "package_name": package_name,
                 "file_name": file_name,
                 "installer": installer,
-                "timestamp": timestamp,
-                "first_install_time": first_install,
-                "last_update_time": last_update,
-                "uid": uid,
                 "disabled": False,
                 "system": False,
                 "third_party": False,
                 "files": package_files,
-            })
+            }
+
+            dumpsys_package = self._adb_command(f"dumpsys package {package_name}")
+            package_details = self.parse_package_for_details(dumpsys_package)
+            new_package.update(package_details)
+
+            self.results.append(new_package)
 
         cmds = [
             {"field": "disabled", "arg": "-d"},
@@ -150,7 +206,7 @@ class Packages(AndroidExtraction):
         ]
         for cmd in cmds:
             output = self._adb_command(f"pm list packages {cmd['arg']}")
-            for line in output.split("\n"):
+            for line in output.splitlines():
                 line = line.strip()
                 if not line.startswith("package:"):
                     continue
@@ -160,6 +216,19 @@ class Packages(AndroidExtraction):
                     if result["package_name"] == package_name:
                         self.results[i][cmd["field"]] = True
 
+        for result in self.results:
+            if not result["third_party"]:
+                continue
+
+            dangerous_permissions_count = 0
+            for perm in result["requested_permissions"]:
+                if perm in DANGEROUS_PERMISSIONS:
+                    dangerous_permissions_count += 1
+
+            if dangerous_permissions_count >= DANGEROUS_PERMISSIONS_THRESHOLD:
+                self.log.info("Third-party package \"%s\" requested %d potentially dangerous permissions",
+                              result["package_name"], dangerous_permissions_count)
+
         packages_to_lookup = []
         for result in self.results:
             if result["system"]:

mvt/android/modules/adb/processes.py

@@ -19,12 +19,22 @@ class Processes(AndroidExtraction):
                          output_folder=output_folder, fast_mode=fast_mode,
                          log=log, results=results)
 
+    def check_indicators(self):
+        if not self.indicators:
+            return
+
+        for result in self.results:
+            ioc = self.indicators.check_app_id(result.get("name", ""))
+            if ioc:
+                result["matched_indicator"] = ioc
+                self.detected.append(result)
+
     def run(self):
         self._adb_connect()
 
         output = self._adb_command("ps -e")
 
-        for line in output.split("\n")[1:]:
+        for line in output.splitlines()[1:]:
             line = line.strip()
             if line == "":
                 continue

mvt/android/modules/adb/root_binaries.py

@@ -25,7 +25,7 @@ class RootBinaries(AndroidExtraction):
     def run(self):
         root_binaries_path = os.path.join("..", "..", "data", "root_binaries.txt")
         root_binaries_string = pkg_resources.resource_string(__name__, root_binaries_path)
-        root_binaries = root_binaries_string.decode("utf-8").split("\n")
+        root_binaries = root_binaries_string.decode("utf-8").splitlines()
 
         self._adb_connect()
 

mvt/android/modules/adb/selinux_status.py

@@ -0,0 +1,40 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2022 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
+
+import pkg_resources
+
+from .base import AndroidExtraction
+
+log = logging.getLogger(__name__)
+
+
+class SELinuxStatus(AndroidExtraction):
+    """This module checks if SELinux is being enforced."""
+
+    slug = "selinux_status"
+
+    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)
+
+        self.results = {} if not results else results
+
+    def run(self):
+        self._adb_connect()
+        output = self._adb_command("getenforce")
+        self._adb_disconnect()
+
+        status = output.lower().strip()
+        self.results["status"] = status
+
+        if status == "enforcing":
+            self.log.info("SELinux is being regularly enforced")
+        else:
+            self.log.warning("SELinux status is \"%s\"!", status)

mvt/android/modules/adb/sms.py

@@ -3,10 +3,15 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
+import base64
+import getpass
 import logging
 import os
 import sqlite3
 
+from mvt.android.parsers.backup import (AndroidBackupParsingError,
+                                        parse_tar_for_sms)
+from mvt.common.module import InsufficientPrivileges
 from mvt.common.utils import check_for_links, convert_timestamp_to_iso
 
 from .base import AndroidExtraction
@@ -16,11 +21,11 @@ log = logging.getLogger(__name__)
 SMS_BUGLE_PATH = "data/data/com.google.android.apps.messaging/databases/bugle_db"
 SMS_BUGLE_QUERY = """
 SELECT
-    ppl.normalized_destination AS number,
+    ppl.normalized_destination AS address,
     p.timestamp AS timestamp,
 CASE WHEN m.sender_id IN
 (SELECT _id FROM participants WHERE contact_id=-1)
-THEN 2 ELSE 1 END incoming, p.text AS text
+THEN 2 ELSE 1 END incoming, p.text AS body
 FROM messages m, conversations c, parts p,
         participants ppl, conversation_participants cp
 WHERE (m.conversation_id = c._id)
@@ -32,10 +37,10 @@ WHERE (m.conversation_id = c._id)
 SMS_MMSSMS_PATH = "data/data/com.android.providers.telephony/databases/mmssms.db"
 SMS_MMSMS_QUERY = """
 SELECT
-    address AS number,
+    address AS address,
     date_sent AS timestamp,
     type as incoming,
-    body AS text
+    body AS body
 FROM sms;
 """
 
@@ -50,12 +55,12 @@ class SMS(AndroidExtraction):
                          log=log, results=results)
 
     def serialize(self, record):
-        text = record["text"].replace("\n", "\\n")
+        body = record["body"].replace("\n", "\\n")
         return {
             "timestamp": record["isodate"],
             "module": self.__class__.__name__,
             "event": f"sms_{record['direction']}",
-            "data": f"{record['number']}: \"{text}\""
+            "data": f"{record['address']}: \"{body}\""
         }
 
     def check_indicators(self):
@@ -63,10 +68,11 @@ class SMS(AndroidExtraction):
             return
 
         for message in self.results:
-            if "text" not in message:
+            if "body" not in message:
                 continue
 
-            message_links = check_for_links(message["text"])
+            # FIXME: check links exported from the body previously
+            message_links = check_for_links(message["body"])
             if self.indicators.check_domains(message_links):
                 self.detected.append(message)
 
@@ -79,9 +85,9 @@ class SMS(AndroidExtraction):
         conn = sqlite3.connect(db_path)
         cur = conn.cursor()
 
-        if (self.SMS_DB_TYPE == 1):
+        if self.SMS_DB_TYPE == 1:
             cur.execute(SMS_BUGLE_QUERY)
-        elif (self.SMS_DB_TYPE == 2):
+        elif self.SMS_DB_TYPE == 2:
             cur.execute(SMS_MMSMS_QUERY)
 
         names = [description[0] for description in cur.description]
@@ -96,7 +102,7 @@ class SMS(AndroidExtraction):
 
             # If we find links in the messages or if they are empty we add
             # them to the list of results.
-            if check_for_links(message["text"]) or message["text"].strip() == "":
+            if check_for_links(message["body"]) or message["body"].strip() == "":
                 self.results.append(message)
 
         cur.close()
@@ -104,12 +110,35 @@ class SMS(AndroidExtraction):
 
         log.info("Extracted a total of %d SMS messages containing links", len(self.results))
 
+    def _extract_sms_adb(self):
+        """Use the Android backup command to extract SMS data from the native SMS app
+
+        It is crucial to use the under-documented "-nocompress" flag to disable the non-standard Java compression
+        algorithim. This module only supports an unencrypted ADB backup.
+        """
+        backup_tar = self._generate_backup("com.android.providers.telephony")
+        if not backup_tar:
+            return
+
+        try:
+            self.results = parse_tar_for_sms(backup_tar)
+        except AndroidBackupParsingError:
+            self.log.info("Impossible to read SMS from the Android Backup, please extract the SMS and try extracting it with Android Backup Extractor")
+            return
+
+        log.info("Extracted a total of %d SMS messages containing links", len(self.results))
+
     def run(self):
-        if (self._adb_check_file_exists(os.path.join("/", SMS_BUGLE_PATH))):
-            self.SMS_DB_TYPE = 1
-            self._adb_process_file(os.path.join("/", SMS_BUGLE_PATH), self._parse_db)
-        elif (self._adb_check_file_exists(os.path.join("/", SMS_MMSSMS_PATH))):
-            self.SMS_DB_TYPE = 2
-            self._adb_process_file(os.path.join("/", SMS_MMSSMS_PATH), self._parse_db)
-        else:
-            self.log.error("No SMS database found")
+        try:
+            if (self._adb_check_file_exists(os.path.join("/", SMS_BUGLE_PATH))):
+                self.SMS_DB_TYPE = 1
+                self._adb_process_file(os.path.join("/", SMS_BUGLE_PATH), self._parse_db)
+            elif (self._adb_check_file_exists(os.path.join("/", SMS_MMSSMS_PATH))):
+                self.SMS_DB_TYPE = 2
+                self._adb_process_file(os.path.join("/", SMS_MMSSMS_PATH), self._parse_db)
+            return
+        except InsufficientPrivileges:
+            pass
+
+        self.log.warn("No SMS database found. Trying extraction of SMS data using Android backup feature.")
+        self._extract_sms_adb()

mvt/android/modules/adb/whatsapp.py

@@ -85,4 +85,7 @@ class Whatsapp(AndroidExtraction):
         self.results = messages
 
     def run(self):
-        self._adb_process_file(os.path.join("/", WHATSAPP_PATH), self._parse_db)
+        try:
+            self._adb_process_file(os.path.join("/", WHATSAPP_PATH), self._parse_db)
+        except Exception as e:
+            self.log.error(e)

mvt/android/modules/backup/base.py

@@ -0,0 +1,46 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2022 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 fnmatch
+import os
+
+from mvt.common.module import MVTModule
+
+
+class BackupExtraction(MVTModule):
+    """This class provides a base for all backup extractios modules"""
+    ab = None
+
+    def from_folder(self, backup_path, files):
+        """
+        Get all the files and list them
+        """
+        self.backup_path = backup_path
+        self.files = files
+
+    def from_ab(self, file_path, tar, files):
+        """
+        Extract the files
+        """
+        self.ab = file_path
+        self.tar = tar
+        self.files = files
+
+    def _get_files_by_pattern(self, pattern):
+        return fnmatch.filter(self.files, pattern)
+
+    def _get_file_content(self, file_path):
+        if self.ab:
+            try:
+                member = self.tar.getmember(file_path)
+            except KeyError:
+                return None
+            handle = self.tar.extractfile(member)
+        else:
+            handle = open(os.path.join(self.backup_path, file_path), "rb")
+
+        data = handle.read()
+        handle.close()
+        return data

mvt/android/modules/backup/sms.py

@@ -3,21 +3,18 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
-import json
-import os
-import zlib
-
-from mvt.common.module import MVTModule
+from mvt.android.modules.backup.base import BackupExtraction
+from mvt.android.parsers.backup import parse_sms_file
 from mvt.common.utils import check_for_links
 
 
-class SMS(MVTModule):
-
+class SMS(BackupExtraction):
     def __init__(self, file_path=None, base_folder=None, output_folder=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)
+        self.results = []
 
     def check_indicators(self):
         if not self.indicators:
@@ -27,38 +24,13 @@ class SMS(MVTModule):
             if "body" not in message:
                 continue
 
-            message_links = check_for_links(message["body"])
-            if self.indicators.check_domains(message_links):
+            if self.indicators.check_domains(message["links"]):
                 self.detected.append(message)
 
-    def _process_sms_file(self, file_path):
-        self.log.info("Processing SMS backup file at %s", file_path)
-
-        with open(file_path, "rb") as handle:
-            data = zlib.decompress(handle.read())
-            json_data = json.loads(data)
-
-        for entry in json_data:
-            message_links = check_for_links(entry["body"])
-
-            # If we find links in the messages or if they are empty we add them to the list.
-            if message_links or entry["body"].strip() == "":
-                self.results.append(entry)
-
     def run(self):
-        app_folder = os.path.join(self.base_folder,
-                                  "apps",
-                                  "com.android.providers.telephony",
-                                  "d_f")
-        if not os.path.exists(app_folder):
-            raise FileNotFoundError("Unable to find the SMS backup folder")
-
-        for file_name in os.listdir(app_folder):
-            if not file_name.endswith("_sms_backup"):
-                continue
-
-            file_path = os.path.join(app_folder, file_name)
-            self._process_sms_file(file_path)
-
+        for file in self._get_files_by_pattern("apps/com.android.providers.telephony/d_f/*_sms_backup"):
+            self.log.info("Processing SMS backup file at %s", file)
+            data = self._get_file_content(file)
+            self.results.extend(parse_sms_file(data))
         self.log.info("Extracted a total of %d SMS messages containing links",
                       len(self.results))

mvt/android/modules/bugreport/__init__.py

@@ -0,0 +1,16 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2022 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/
+
+from .accessibility import Accessibility
+from .activities import Activities
+from .battery_daily import BatteryDaily
+from .battery_history import BatteryHistory
+from .dbinfo import DBInfo
+from .getprop import Getprop
+from .packages import Packages
+from .receivers import Receivers
+
+BUGREPORT_MODULES = [Accessibility, Activities, BatteryDaily, BatteryHistory,
+                     DBInfo, Getprop, Packages, Receivers]

mvt/android/modules/bugreport/accessibility.py

@@ -0,0 +1,60 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2022 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.android.parsers import parse_dumpsys_accessibility
+
+from .base import BugReportModule
+
+log = logging.getLogger(__name__)
+
+
+class Accessibility(BugReportModule):
+    """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 check_indicators(self):
+        if not self.indicators:
+            return
+
+        for result in self.results:
+            ioc = self.indicators.check_app_id(result["package_name"])
+            if ioc:
+                result["matched_indicator"] = ioc
+                self.detected.append(result)
+                continue
+
+    def run(self):
+        content = self._get_dumpstate_file()
+        if not content:
+            self.log.error("Unable to find dumpstate file. Did you provide a valid bug report archive?")
+            return
+
+        lines = []
+        in_accessibility = False
+        for line in content.decode(errors="ignore").splitlines():
+            if line.strip() == "DUMP OF SERVICE accessibility:":
+                in_accessibility = True
+                continue
+
+            if not in_accessibility:
+                continue
+
+            if line.strip().startswith("------------------------------------------------------------------------------"):
+                break
+
+            lines.append(line)
+
+        self.results = parse_dumpsys_accessibility("\n".join(lines))
+        for result in self.results:
+            log.info("Found installed accessibility service \"%s\"", result.get("service"))
+
+        self.log.info("Identified a total of %d accessibility services", len(self.results))

mvt/android/modules/bugreport/activities.py

@@ -0,0 +1,61 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2022 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.android.parsers import parse_dumpsys_activity_resolver_table
+
+from .base import BugReportModule
+
+log = logging.getLogger(__name__)
+
+
+class Activities(BugReportModule):
+    """This module extracts details on receivers for risky activities."""
+
+    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)
+
+        self.results = results if results else {}
+
+    def check_indicators(self):
+        if not self.indicators:
+            return
+
+        for intent, activities in self.results.items():
+            for activity in activities:
+                ioc = self.indicators.check_app_id(activity["package_name"])
+                if ioc:
+                    activity["matched_indicator"] = ioc
+                    self.detected.append({intent: activity})
+                    continue
+
+    def run(self):
+        content = self._get_dumpstate_file()
+        if not content:
+            self.log.error("Unable to find dumpstate file. Did you provide a valid bug report archive?")
+            return
+
+        lines = []
+        in_package = False
+        for line in content.decode(errors="ignore").splitlines():
+            if line.strip() == "DUMP OF SERVICE package:":
+                in_package = True
+                continue
+
+            if not in_package:
+                continue
+
+            if line.strip().startswith("------------------------------------------------------------------------------"):
+                break
+
+            lines.append(line)
+
+        self.results = parse_dumpsys_activity_resolver_table("\n".join(lines))
+
+        self.log.info("Extracted activities for %d intents", len(self.results))

mvt/android/modules/bugreport/base.py

@@ -0,0 +1,70 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2022 Claudio Guarnieri.
+# See the file 'LICENSE' for usage and copying permissions, or find a copy at
+#   https://github.com/mvt-project/mvt/blob/main/LICENSE
+
+import fnmatch
+import logging
+import os
+
+from mvt.common.module import MVTModule
+
+log = logging.getLogger(__name__)
+
+
+class BugReportModule(MVTModule):
+    """This class provides a base for all Android Bug Report modules."""
+
+    zip_archive = None
+
+    def from_folder(self, extract_path, extract_files):
+        self.extract_path = extract_path
+        self.extract_files = extract_files
+
+    def from_zip(self, zip_archive, zip_files):
+        self.zip_archive = zip_archive
+        self.zip_files = zip_files
+
+    def _get_files_by_pattern(self, pattern):
+        file_names = []
+        if self.zip_archive:
+            for zip_file in self.zip_files:
+                file_names.append(zip_file)
+        else:
+            file_names = self.extract_files
+
+        return fnmatch.filter(file_names, pattern)
+
+    def _get_files_by_patterns(self, patterns):
+        for pattern in patterns:
+            matches = self._get_files_by_pattern(pattern)
+            if matches:
+                return matches
+
+    def _get_file_content(self, file_path):
+        if self.zip_archive:
+            handle = self.zip_archive.open(file_path)
+        else:
+            handle = open(os.path.join(self.extract_path, file_path), "rb")
+
+        data = handle.read()
+        handle.close()
+
+        return data
+
+    def _get_dumpstate_file(self):
+        main = self._get_files_by_pattern("main_entry.txt")
+        if main:
+            main_content = self._get_file_content(main[0])
+            try:
+                return self._get_file_content(main_content.decode().strip())
+            except KeyError:
+                return None
+        else:
+            dumpstate_logs = self._get_files_by_pattern("dumpState_*.log")
+            if not dumpstate_logs:
+                return None
+
+            return self._get_file_content(dumpstate_logs[0])
+
+        return None

mvt/android/modules/bugreport/battery_daily.py

@@ -0,0 +1,76 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2022 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.android.parsers import parse_dumpsys_battery_daily
+
+from .base import BugReportModule
+
+log = logging.getLogger(__name__)
+
+
+class BatteryDaily(BugReportModule):
+    """This module extracts records from battery daily updates."""
+
+    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 serialize(self, record):
+        return {
+            "timestamp": record["from"],
+            "module": self.__class__.__name__,
+            "event": "battery_daily",
+            "data": f"Recorded update of package {record['package_name']} with vers {record['vers']}"
+        }
+
+    def check_indicators(self):
+        if not self.indicators:
+            return
+
+        for result in self.results:
+            ioc = self.indicators.check_app_id(result["package_name"])
+            if ioc:
+                result["matched_indicator"] = ioc
+                self.detected.append(result)
+                continue
+
+    def run(self):
+        content = self._get_dumpstate_file()
+        if not content:
+            self.log.error("Unable to find dumpstate file. Did you provide a valid bug report archive?")
+            return
+
+        lines = []
+        in_batterystats = False
+        in_daily = False
+        for line in content.decode(errors="ignore").splitlines():
+            if line.strip() == "DUMP OF SERVICE batterystats:":
+                in_batterystats = True
+                continue
+
+            if not in_batterystats:
+                continue
+
+            if line.strip() == "Daily stats:":
+                lines.append(line)
+                in_daily = True
+                continue
+
+            if not in_daily:
+                continue
+
+            if line.strip() == "":
+                break
+
+            lines.append(line)
+
+        self.results = parse_dumpsys_battery_daily("\n".join(lines))
+
+        self.log.info("Extracted a total of %d battery daily stats",
+                      len(self.results))

mvt/android/modules/bugreport/battery_history.py

@@ -0,0 +1,60 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2022 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.android.parsers import parse_dumpsys_battery_history
+
+from .base import BugReportModule
+
+log = logging.getLogger(__name__)
+
+
+class BatteryHistory(BugReportModule):
+    """This module extracts records from battery daily updates."""
+
+    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 check_indicators(self):
+        if not self.indicators:
+            return
+
+        for result in self.results:
+            ioc = self.indicators.check_app_id(result["package_name"])
+            if ioc:
+                result["matched_indicator"] = ioc
+                self.detected.append(result)
+                continue
+
+    def run(self):
+        content = self._get_dumpstate_file()
+        if not content:
+            self.log.error("Unable to find dumpstate file. Did you provide a valid bug report archive?")
+            return
+
+        lines = []
+        in_history = False
+        for line in content.decode(errors="ignore").splitlines():
+            if line.strip().startswith("Battery History "):
+                lines.append(line)
+                in_history = True
+                continue
+
+            if not in_history:
+                continue
+
+            if line.strip() == "":
+                break
+
+            lines.append(line)
+
+        self.results = parse_dumpsys_battery_history("\n".join(lines))
+
+        self.log.info("Extracted a total of %d battery history records",
+                      len(self.results))

mvt/android/modules/bugreport/dbinfo.py

@@ -0,0 +1,63 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2022 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.android.parsers import parse_dumpsys_dbinfo
+
+from .base import BugReportModule
+
+log = logging.getLogger(__name__)
+
+
+class DBInfo(BugReportModule):
+    """This module extracts records from battery daily updates."""
+
+    slug = "dbinfo"
+
+    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 check_indicators(self):
+        if not self.indicators:
+            return
+
+        for result in self.results:
+            path = result.get("path", "")
+            for part in path.split("/"):
+                ioc = self.indicators.check_app_id(part)
+                if ioc:
+                    result["matched_indicator"] = ioc
+                    self.detected.append(result)
+                    continue
+
+    def run(self):
+        content = self._get_dumpstate_file()
+        if not content:
+            self.log.error("Unable to find dumpstate file. Did you provide a valid bug report archive?")
+            return
+
+        in_dbinfo = False
+        lines = []
+        for line in content.decode(errors="ignore").splitlines():
+            if line.strip() == "DUMP OF SERVICE dbinfo:":
+                in_dbinfo = True
+                continue
+
+            if not in_dbinfo:
+                continue
+
+            if line.strip().startswith("------------------------------------------------------------------------------"):
+                break
+
+            lines.append(line)
+
+        self.results = parse_dumpsys_dbinfo("\n".join(lines))
+
+        self.log.info("Extracted a total of %d database connection pool records",
+                      len(self.results))

mvt/android/modules/bugreport/getprop.py

@@ -0,0 +1,59 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2022 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 re
+from datetime import datetime, timedelta
+
+from mvt.android.parsers import parse_getprop
+
+from .base import BugReportModule
+
+log = logging.getLogger(__name__)
+
+
+class Getprop(BugReportModule):
+    """This module extracts device properties from getprop command."""
+
+    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)
+
+        self.results = {} if not results else results
+
+    def run(self):
+        content = self._get_dumpstate_file()
+        if not content:
+            self.log.error("Unable to find dumpstate file. Did you provide a valid bug report archive?")
+            return
+
+        lines = []
+        in_getprop = False
+        for line in content.decode(errors="ignore").splitlines():
+            if line.strip() == "------ SYSTEM PROPERTIES (getprop) ------":
+                in_getprop = True
+                continue
+
+            if not in_getprop:
+                continue
+
+            if line.strip() == "------":
+                break
+
+            lines.append(line)
+
+        self.results = parse_getprop("\n".join(lines))
+
+        # Alert if phone is outdated.
+        security_patch = self.results.get("ro.build.version.security_patch", "")
+        if security_patch:
+            patch_date = datetime.strptime(security_patch, "%Y-%m-%d")
+            if (datetime.now() - patch_date) > timedelta(days=6*30):
+                self.log.warning("This phone has not received security updates for more than "
+                                 "six months (last update: %s)", security_patch)
+
+        self.log.info("Extracted %d Android system properties", len(self.results))

mvt/android/modules/bugreport/packages.py

@@ -0,0 +1,117 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2022 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 re
+
+from mvt.android.modules.adb.packages import Packages as PCK
+
+from .base import BugReportModule
+
+log = logging.getLogger(__name__)
+
+
+class Packages(BugReportModule):
+    """This module extracts details on receivers for risky activities."""
+
+    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 serialize(self, record):
+        records = []
+
+        timestamps = [
+            {"event": "package_install", "timestamp": record["timestamp"]},
+            {"event": "package_first_install", "timestamp": record["first_install_time"]},
+            {"event": "package_last_update", "timestamp": record["last_update_time"]},
+        ]
+
+        for ts in timestamps:
+            records.append({
+                "timestamp": ts["timestamp"],
+                "module": self.__class__.__name__,
+                "event": ts["event"],
+                "data": f"Install or update of package {record['package_name']}",
+            })
+
+        return records
+
+    def check_indicators(self):
+        if not self.indicators:
+            return
+
+        for result in self.results:
+            ioc = self.indicators.check_app_id(result["package_name"])
+            if ioc:
+                result["matched_indicator"] = ioc
+                self.detected.append(result)
+                continue
+
+    @staticmethod
+    def parse_packages_list(output):
+        pkg_rxp = re.compile(r"  Package \[(.+?)\].*")
+
+        results = []
+        package_name = None
+        package = {}
+        lines = []
+        for line in output.splitlines():
+            if line.startswith("  Package ["):
+                if len(lines) > 0:
+                    details = PCK.parse_package_for_details("\n".join(lines))
+                    package.update(details)
+                    results.append(package)
+                    package = {}
+
+                matches = pkg_rxp.findall(line)
+                if not matches:
+                    continue
+
+                package_name = matches[0]
+                package["package_name"] = package_name
+                continue
+
+            if not package_name:
+                continue
+
+            lines.append(line)
+
+        return results
+
+    def run(self):
+        content = self._get_dumpstate_file()
+        if not content:
+            self.log.error("Unable to find dumpstate file. Did you provide a valid bug report archive?")
+            return
+
+        in_package = False
+        in_packages_list = False
+        lines = []
+        for line in content.decode(errors="ignore").splitlines():
+            if line.strip() == "DUMP OF SERVICE package:":
+                in_package = True
+                continue
+
+            if not in_package:
+                continue
+
+            if line.strip() == "Packages:":
+                in_packages_list = True
+                continue
+
+            if not in_packages_list:
+                continue
+
+            if line.strip() == "":
+                break
+
+            lines.append(line)
+
+        self.results = self.parse_packages_list("\n".join(lines))
+
+        self.log.info("Extracted details on %d packages", len(self.results))

mvt/android/modules/bugreport/receivers.py

@@ -0,0 +1,83 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2022 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.android.parsers import parse_dumpsys_receiver_resolver_table
+
+from .base import BugReportModule
+
+log = logging.getLogger(__name__)
+
+INTENT_NEW_OUTGOING_SMS = "android.provider.Telephony.NEW_OUTGOING_SMS"
+INTENT_SMS_RECEIVED = "android.provider.Telephony.SMS_RECEIVED"
+INTENT_DATA_SMS_RECEIVED = "android.intent.action.DATA_SMS_RECEIVED"
+INTENT_PHONE_STATE = "android.intent.action.PHONE_STATE"
+INTENT_NEW_OUTGOING_CALL = "android.intent.action.NEW_OUTGOING_CALL"
+
+
+class Receivers(BugReportModule):
+    """This module extracts details on receivers for risky activities."""
+
+    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)
+
+        self.results = results if results else {}
+
+    def check_indicators(self):
+        if not self.indicators:
+            return
+
+        for intent, receivers in self.results.items():
+            for receiver in receivers:
+                if intent == INTENT_NEW_OUTGOING_SMS:
+                    self.log.info("Found a receiver to intercept outgoing SMS messages: \"%s\"",
+                                  receiver["receiver"])
+                elif intent == INTENT_SMS_RECEIVED:
+                    self.log.info("Found a receiver to intercept incoming SMS messages: \"%s\"",
+                                  receiver["receiver"])
+                elif intent == INTENT_DATA_SMS_RECEIVED:
+                    self.log.info("Found a receiver to intercept incoming data SMS message: \"%s\"",
+                                  receiver["receiver"])
+                elif intent == INTENT_PHONE_STATE:
+                    self.log.info("Found a receiver monitoring telephony state/incoming calls: \"%s\"",
+                                  receiver["receiver"])
+                elif intent == INTENT_NEW_OUTGOING_CALL:
+                    self.log.info("Found a receiver monitoring outgoing calls: \"%s\"",
+                                  receiver["receiver"])
+
+            ioc = self.indicators.check_app_id(receiver["package_name"])
+            if ioc:
+                receiver["matched_indicator"] = ioc
+                self.detected.append({intent: receiver})
+                continue
+
+    def run(self):
+        content = self._get_dumpstate_file()
+        if not content:
+            self.log.error("Unable to find dumpstate file. Did you provide a valid bug report archive?")
+            return
+
+        in_receivers = False
+        lines = []
+        for line in content.decode(errors="ignore").splitlines():
+            if line.strip() == "DUMP OF SERVICE package:":
+                in_receivers = True
+                continue
+
+            if not in_receivers:
+                continue
+
+            if line.strip().startswith("------------------------------------------------------------------------------"):
+                break
+
+            lines.append(line)
+
+        self.results = parse_dumpsys_receiver_resolver_table("\n".join(lines))
+
+        self.log.info("Extracted receivers for %d intents", len(self.results))

mvt/android/parsers/__init__.py

@@ -0,0 +1,11 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2022 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/
+
+from .dumpsys import (parse_dumpsys_accessibility,
+                      parse_dumpsys_activity_resolver_table,
+                      parse_dumpsys_battery_daily,
+                      parse_dumpsys_battery_history, parse_dumpsys_dbinfo,
+                      parse_dumpsys_receiver_resolver_table)
+from .getprop import parse_getprop

mvt/android/parsers/backup.py

@@ -0,0 +1,202 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2022 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 datetime
+import io
+import json
+import tarfile
+import zlib
+
+from cryptography.hazmat.primitives import hashes, padding
+from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
+from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
+
+from mvt.common.utils import check_for_links, convert_timestamp_to_iso
+
+PBKDF2_KEY_SIZE = 32
+
+
+class AndroidBackupParsingError(Exception):
+    """Exception raised file parsing an android backup file"""
+
+
+class AndroidBackupNotImplemented(AndroidBackupParsingError):
+    pass
+
+
+class InvalidBackupPassword(AndroidBackupParsingError):
+    pass
+
+
+def to_utf8_bytes(input_bytes):
+    output = []
+    for byte in input_bytes:
+        if byte < ord(b'\x80'):
+            output.append(byte)
+        else:
+            output.append(ord('\xef') | (byte >> 12))
+            output.append(ord('\xbc') | ((byte >> 6) & ord('\x3f')))
+            output.append(ord('\x80') | (byte & ord('\x3f')))
+    return bytes(output)
+
+
+def parse_ab_header(data):
+    """
+    Parse the header of an Android Backup file
+    Returns a dict {'backup': True, 'compression': False,
+        'encryption': "none", 'version': 4}
+    """
+    if data.startswith(b"ANDROID BACKUP"):
+        [magic_header, version, is_compressed, encryption, tar_data] = data.split(b"\n", 4)
+        return {
+            "backup": True,
+            "compression": (is_compressed == b"1"),
+            "version": int(version),
+            "encryption": encryption.decode("utf-8")
+        }
+
+    return {
+        "backup": False,
+        "compression": None,
+        "version": None,
+        "encryption": None
+    }
+
+
+def decrypt_master_key(password, user_salt, user_iv, pbkdf2_rounds, master_key_blob, format_version, checksum_salt):
+    """Generate AES key from user password uisng PBKDF2
+
+    The backup master key is extracted from the master key blog after decryption.
+    """
+    # Derive key from password using PBKDF2
+    kdf = PBKDF2HMAC(algorithm=hashes.SHA1(), length=32, salt=user_salt, iterations=pbkdf2_rounds)
+    key = kdf.derive(password.encode("utf-8"))
+
+    # Decrypt master key blob
+    cipher = Cipher(algorithms.AES(key), modes.CBC(user_iv))
+    decryptor = cipher.decryptor()
+    try:
+        decryted_master_key_blob = decryptor.update(master_key_blob) + decryptor.finalize()
+
+        # Extract key and IV from decrypted blob.
+        key_blob = io.BytesIO(decryted_master_key_blob)
+        master_iv_length = ord(key_blob.read(1))
+        master_iv = key_blob.read(master_iv_length)
+
+        master_key_length = ord(key_blob.read(1))
+        master_key = key_blob.read(master_key_length)
+
+        master_key_checksum_length = ord(key_blob.read(1))
+        master_key_checksum = key_blob.read(master_key_checksum_length)
+    except TypeError:
+        raise InvalidBackupPassword()
+
+    # Handle quirky encoding of master key bytes in Android original Java crypto code
+    if format_version > 1:
+        hmac_mk = to_utf8_bytes(master_key)
+    else:
+        hmac_mk = master_key
+
+    # Derive checksum to confirm successful backup decryption.
+    kdf = PBKDF2HMAC(algorithm=hashes.SHA1(), length=32, salt=checksum_salt, iterations=pbkdf2_rounds)
+    calculated_checksum = kdf.derive(hmac_mk)
+
+    if master_key_checksum != calculated_checksum:
+        raise InvalidBackupPassword()
+
+    return master_key, master_iv
+
+
+def decrypt_backup_data(encrypted_backup, password, encryption_algo, format_version):
+    """
+    Generate encryption keyffrom password and do decryption
+    """
+    if encryption_algo != b"AES-256":
+        raise AndroidBackupNotImplemented("Encryption Algorithm not implemented")
+
+    if password is None:
+        raise InvalidBackupPassword()
+
+    [user_salt, checksum_salt, pbkdf2_rounds, user_iv, master_key_blob, encrypted_data] = encrypted_backup.split(b"\n", 5)
+    user_salt = bytes.fromhex(user_salt.decode("utf-8"))
+    checksum_salt = bytes.fromhex(checksum_salt.decode("utf-8"))
+    pbkdf2_rounds = int(pbkdf2_rounds)
+    user_iv = bytes.fromhex(user_iv.decode("utf-8"))
+    master_key_blob = bytes.fromhex(master_key_blob.decode("utf-8"))
+
+    # Derive decryption master key from password
+    master_key, master_iv = decrypt_master_key(password=password, user_salt=user_salt, user_iv=user_iv,
+                                               pbkdf2_rounds=pbkdf2_rounds, master_key_blob=master_key_blob,
+                                               format_version=format_version, checksum_salt=checksum_salt)
+
+    # Decrypt and unpad backup data using derivied key
+    cipher = Cipher(algorithms.AES(master_key), modes.CBC(master_iv))
+    decryptor = cipher.decryptor()
+    decrypted_tar = decryptor.update(encrypted_data) + decryptor.finalize()
+
+    unpadder = padding.PKCS7(128).unpadder()
+    return unpadder.update(decrypted_tar)
+
+
+def parse_backup_file(data, password=None):
+    """
+    Parse an ab file, returns a tar file
+    """
+    if not data.startswith(b"ANDROID BACKUP"):
+        raise AndroidBackupParsingError("Invalid file header")
+
+    [magic_header, version, is_compressed, encryption_algo, tar_data] = data.split(b"\n", 4)
+    version = int(version)
+    is_compressed = int(is_compressed)
+
+    if encryption_algo != b"none":
+        tar_data = decrypt_backup_data(tar_data, password, encryption_algo, format_version=version)
+
+    if is_compressed:
+        try:
+            tar_data = zlib.decompress(tar_data)
+        except zlib.error:
+            raise AndroidBackupParsingError("Impossible to decompress the backup file")
+
+    return tar_data
+
+
+def parse_tar_for_sms(data):
+    """
+    Extract SMS from a tar backup archive
+    Returns an array of SMS
+    """
+    dbytes = io.BytesIO(data)
+    tar = tarfile.open(fileobj=dbytes)
+    try:
+        member = tar.getmember("apps/com.android.providers.telephony/d_f/000000_sms_backup")
+    except KeyError:
+        return []
+
+    dhandler = tar.extractfile(member)
+    return parse_sms_file(dhandler.read())
+
+
+def parse_sms_file(data):
+    """
+    Parse an SMS file extracted from a folder
+    Returns a list of SMS entries
+    """
+    res = []
+    data = zlib.decompress(data)
+    json_data = json.loads(data)
+
+    for entry in json_data:
+        message_links = check_for_links(entry["body"])
+        utc_timestamp = datetime.datetime.utcfromtimestamp(int(entry["date"]) / 1000)
+        entry["isodate"] = convert_timestamp_to_iso(utc_timestamp)
+        entry["direction"] = ("sent" if int(entry["date_sent"]) else "received")
+
+        # If we find links in the messages or if they are empty we add them to the list.
+        if message_links or entry["body"].strip() == "":
+            entry["links"] = message_links
+            res.append(entry)
+
+    return res

mvt/android/parsers/dumpsys.py

@@ -0,0 +1,287 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2022 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 re
+
+
+def parse_dumpsys_accessibility(output):
+    results = []
+
+    in_services = False
+    for line in output.splitlines():
+        if line.strip().startswith("installed services:"):
+            in_services = True
+            continue
+
+        if not in_services:
+            continue
+
+        if line.strip() == "}":
+            break
+
+        service = line.split(":")[1].strip()
+
+        results.append({
+            "package_name": service.split("/")[0],
+            "service": service,
+        })
+
+    return results
+
+
+def parse_dumpsys_activity_resolver_table(output):
+    results = {}
+
+    in_activity_resolver_table = False
+    in_non_data_actions = False
+    intent = None
+    for line in output.splitlines():
+        if line.startswith("Activity Resolver Table:"):
+            in_activity_resolver_table = True
+            continue
+
+        if not in_activity_resolver_table:
+            continue
+
+        if line.startswith("  Non-Data Actions:"):
+            in_non_data_actions = True
+            continue
+
+        if not in_non_data_actions:
+            continue
+
+        # If we hit an empty line, the Non-Data Actions section should be
+        # finished.
+        if line.strip() == "":
+            break
+
+        # We detect the action name.
+        if line.startswith(" " * 6) and not line.startswith(" " * 8) and ":" in line:
+            intent = line.strip().replace(":", "")
+            results[intent] = []
+            continue
+
+        # If we are not in an intent block yet, skip.
+        if not intent:
+            continue
+
+        # If we are in a block but the line does not start with 8 spaces
+        # it means the block ended a new one started, so we reset and
+        # continue.
+        if not line.startswith(" " * 8):
+            intent = None
+            continue
+
+        # If we got this far, we are processing receivers for the
+        # activities we are interested in.
+        activity = line.strip().split(" ")[1]
+        package_name = activity.split("/")[0]
+
+        results[intent].append({
+            "package_name": package_name,
+            "activity": activity,
+        })
+
+    return results
+
+
+def parse_dumpsys_battery_daily(output):
+    results = []
+    daily = None
+    daily_updates = []
+    for line in output.splitlines():
+        if line.startswith("  Daily from "):
+            if len(daily_updates) > 0:
+                results.extend(daily_updates)
+                daily_updates = []
+
+            timeframe = line[13:].strip()
+            date_from, date_to = timeframe.strip(":").split(" to ", 1)
+            daily = {"from": date_from[0:10], "to": date_to[0:10]}
+            continue
+
+        if not daily:
+            continue
+
+        if not line.strip().startswith("Update "):
+            continue
+
+        line = line.strip().replace("Update ", "")
+        package_name, vers = line.split(" ", 1)
+        vers_nr = vers.split("=", 1)[1]
+
+        already_seen = False
+        for update in daily_updates:
+            if package_name == update["package_name"] and vers_nr == update["vers"]:
+                already_seen = True
+                break
+
+        if not already_seen:
+            daily_updates.append({
+                "action": "update",
+                "from": daily["from"],
+                "to": daily["to"],
+                "package_name": package_name,
+                "vers": vers_nr,
+            })
+
+    if len(daily_updates) > 0:
+        results.extend(daily_updates)
+
+    return results
+
+
+def parse_dumpsys_battery_history(output):
+    results = []
+
+    for line in output.splitlines():
+        if line.startswith("Battery History "):
+            continue
+
+        if line.strip() == "":
+            break
+
+        time_elapsed = line.strip().split(" ", 1)[0]
+
+        event = ""
+        if line.find("+job") > 0:
+            event = "start_job"
+            uid = line[line.find("+job")+5:line.find(":")]
+            service = line[line.find(":")+1:].strip('"')
+            package_name = service.split("/")[0]
+        elif line.find("-job") > 0:
+            event = "end_job"
+            uid = line[line.find("-job")+5:line.find(":")]
+            service = line[line.find(":")+1:].strip('"')
+            package_name = service.split("/")[0]
+        elif line.find("+running +wake_lock=") > 0:
+            uid = line[line.find("+running +wake_lock=")+21:line.find(":")]
+            event = "wake"
+            service = line[line.find("*walarm*:")+9:].split(" ")[0].strip('"').strip()
+            if service == "" or "/" not in service:
+                continue
+
+            package_name = service.split("/")[0]
+        else:
+            continue
+
+        results.append({
+            "time_elapsed": time_elapsed,
+            "event": event,
+            "uid": uid,
+            "package_name": package_name,
+            "service": service,
+        })
+
+    return results
+
+
+def parse_dumpsys_dbinfo(output):
+    results = []
+
+    rxp = re.compile(r'.*\[([0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3})\].*\[Pid:\((\d+)\)\](\w+).*sql\=\"(.+?)\"')
+    rxp_no_pid = re.compile(r'.*\[([0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3})\][ ]{1}(\w+).*sql\=\"(.+?)\"')
+
+    pool = None
+    in_operations = False
+    for line in output.splitlines():
+        if line.startswith("Connection pool for "):
+            pool = line.replace("Connection pool for ", "").rstrip(":")
+
+        if not pool:
+            continue
+
+        if line.strip() == "Most recently executed operations:":
+            in_operations = True
+            continue
+
+        if not in_operations:
+            continue
+
+        if not line.startswith("        "):
+            in_operations = False
+            pool = None
+            continue
+
+        matches = rxp.findall(line)
+        if not matches:
+            matches = rxp_no_pid.findall(line)
+            if not matches:
+                continue
+            else:
+                match = matches[0]
+                results.append({
+                    "isodate": match[0],
+                    "action": match[1],
+                    "sql": match[2],
+                    "path": pool,
+                })
+        else:
+            match = matches[0]
+            results.append({
+                "isodate": match[0],
+                "pid": match[1],
+                "action": match[2],
+                "sql": match[3],
+                "path": pool,
+            })
+
+    return results
+
+
+def parse_dumpsys_receiver_resolver_table(output):
+    results = {}
+
+    in_receiver_resolver_table = False
+    in_non_data_actions = False
+    intent = None
+    for line in output.splitlines():
+        if line.startswith("Receiver Resolver Table:"):
+            in_receiver_resolver_table = True
+            continue
+
+        if not in_receiver_resolver_table:
+            continue
+
+        if line.startswith("  Non-Data Actions:"):
+            in_non_data_actions = True
+            continue
+
+        if not in_non_data_actions:
+            continue
+
+        # If we hit an empty line, the Non-Data Actions section should be
+        # finished.
+        if line.strip() == "":
+            break
+
+        # We detect the action name.
+        if line.startswith(" " * 6) and not line.startswith(" " * 8) and ":" in line:
+            intent = line.strip().replace(":", "")
+            results[intent] = []
+            continue
+
+        # If we are not in an intent block yet, skip.
+        if not intent:
+            continue
+
+        # If we are in a block but the line does not start with 8 spaces
+        # it means the block ended a new one started, so we reset and
+        # continue.
+        if not line.startswith(" " * 8):
+            intent = None
+            continue
+
+        # If we got this far, we are processing receivers for the
+        # activities we are interested in.
+        receiver = line.strip().split(" ")[1]
+        package_name = receiver.split("/")[0]
+
+        results[intent].append({
+            "package_name": package_name,
+            "receiver": receiver,
+        })
+
+    return results

mvt/android/parsers/getprop.py

@@ -0,0 +1,26 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2022 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 re
+
+
+def parse_getprop(output):
+    results = {}
+    rxp = re.compile(r"\[(.+?)\]: \[(.+?)\]")
+
+    for line in output.splitlines():
+        line = line.strip()
+        if line == "":
+            continue
+
+        matches = re.findall(rxp, line)
+        if not matches or len(matches[0]) != 2:
+            continue
+
+        key = matches[0][0]
+        value = matches[0][1]
+        results[key] = value
+
+    return results

mvt/common/indicators.py

@@ -20,7 +20,7 @@ class Indicators:
     def __init__(self, log=None):
         self.data_dir = user_data_dir("mvt")
         self.log = log
-        self.ioc_files = []
+        self.ioc_collections = []
         self.total_ioc_count = 0
 
     def _load_downloaded_indicators(self):
@@ -33,7 +33,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 a STIX files.
         """
         if "MVT_STIX2" not in os.environ:
             return
@@ -43,14 +43,17 @@ class Indicators:
             if os.path.isfile(path):
                 self.parse_stix2(path)
             else:
-                self.log.info("Invalid STIX2 path %s in MVT_STIX2 environment variable", path)
+                self.log.error("Path specified with env MVT_STIX2 is not a valid file: %s",
+                               path)
 
-    def _generate_indicators_file(self):
+    def _new_collection(self, cid="", name="", description="", file_name="",
+                        file_path=""):
         return {
-            "name": "",
-            "description": "",
-            "file_name": "",
-            "file_path": "",
+            "id": cid,
+            "name": name,
+            "description": description,
+            "stix2_file_name": file_name,
+            "stix2_file_path": file_path,
             "domains": [],
             "processes": [],
             "emails": [],
@@ -62,10 +65,11 @@ class Indicators:
             "count": 0,
         }
 
-    def _add_indicator(self, ioc, ioc_file, iocs_list):
-        if ioc not in iocs_list:
-            iocs_list.append(ioc)
-            ioc_file["count"] += 1
+    def _add_indicator(self, ioc, ioc_coll, ioc_coll_list):
+        ioc = ioc.strip("'")
+        if ioc not in ioc_coll_list:
+            ioc_coll_list.append(ioc)
+            ioc_coll["count"] += 1
             self.total_ioc_count += 1
 
     def parse_stix2(self, file_path):
@@ -77,69 +81,102 @@ 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", encoding="utf-8") as handle:
             try:
                 data = json.load(handle)
             except json.decoder.JSONDecodeError:
-                self.log.critical("Unable to parse STIX2 indicator file. The file is malformed or in the wrong format.")
+                self.log.critical("Unable to parse STIX2 indicator file. "
+                                  "The file is corrupted or in the wrong format!")
                 return
 
+        malware = {}
+        indicators = []
+        relationships = []
         for entry in data.get("objects", []):
             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
+                malware[entry["id"]] = {
+                    "name": entry["name"],
+                    "description": entry["description"],
+                }
+            elif entry_type == "indicator":
+                indicators.append(entry)
+            elif entry_type == "relationship":
+                relationships.append(entry)
 
-            if entry_type != "indicator":
-                continue
+        collections = []
+        for mal_id, mal_values in malware.items():
+            collection = self._new_collection(mal_id, mal_values.get("name"),
+                                              mal_values.get("description"),
+                                              os.path.basename(file_path),
+                                              file_path)
+            collections.append(collection)
 
-            key, value = entry.get("pattern", "").strip("[]").split("=")
-            value = value.strip("'")
+        # We loop through all indicators.
+        for indicator in indicators:
+            malware_id = None
 
-            if key == "domain-name:value":
-                # We force domain names to lower case.
-                self._add_indicator(ioc=value.lower(),
-                                    ioc_file=ioc_file,
-                                    iocs_list=ioc_file["domains"])
-            elif key == "process:name":
-                self._add_indicator(ioc=value,
-                                    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(),
-                                    ioc_file=ioc_file,
-                                    iocs_list=ioc_file["emails"])
-            elif key == "file:name":
-                self._add_indicator(ioc=value,
-                                    ioc_file=ioc_file,
-                                    iocs_list=ioc_file["file_names"])
-            elif key == "file:path":
-                self._add_indicator(ioc=value,
-                                    ioc_file=ioc_file,
-                                    iocs_list=ioc_file["file_paths"])
-            elif key == "file:hashes.sha256":
-                self._add_indicator(ioc=value,
-                                    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"])
+            # We loop through all relationships and find the one pertinent to
+            # the current indicator.
+            for relationship in relationships:
+                if relationship["source_ref"] != indicator["id"]:
+                    continue
 
-        self.log.info("Loaded %d indicators from \"%s\" indicators file",
-                      ioc_file["count"], ioc_file["name"])
+                # Look for a malware definition with the correct identifier.
+                if relationship["target_ref"] in malware.keys():
+                    malware_id = relationship["target_ref"]
+                    break
 
-        self.ioc_files.append(ioc_file)
+            # Now we look for the correct collection matching the malware ID we
+            # got from the relationship.
+            for collection in collections:
+                if collection["id"] != malware_id:
+                    continue
+
+                key, value = indicator.get("pattern", "").strip("[]").split("=")
+
+                if key == "domain-name:value":
+                    # We force domain names to lower case.
+                    self._add_indicator(ioc=value.lower(),
+                                        ioc_coll=collection,
+                                        ioc_coll_list=collection["domains"])
+                elif key == "process:name":
+                    self._add_indicator(ioc=value,
+                                        ioc_coll=collection,
+                                        ioc_coll_list=collection["processes"])
+                elif key == "email-addr:value":
+                    # We force email addresses to lower case.
+                    self._add_indicator(ioc=value.lower(),
+                                        ioc_coll=collection,
+                                        ioc_coll_list=collection["emails"])
+                elif key == "file:name":
+                    self._add_indicator(ioc=value,
+                                        ioc_coll=collection,
+                                        ioc_coll_list=collection["file_names"])
+                elif key == "file:path":
+                    self._add_indicator(ioc=value,
+                                        ioc_coll=collection,
+                                        ioc_coll_list=collection["file_paths"])
+                elif key == "file:hashes.sha256":
+                    self._add_indicator(ioc=value,
+                                        ioc_coll=collection,
+                                        ioc_coll_list=collection["files_sha256"])
+                elif key == "app:id":
+                    self._add_indicator(ioc=value,
+                                        ioc_coll=collection,
+                                        ioc_coll_list=collection["app_ids"])
+                elif key == "configuration-profile:id":
+                    self._add_indicator(ioc=value,
+                                        ioc_coll=collection,
+                                        ioc_coll_list=collection["ios_profile_ids"])
+
+                break
+
+        for coll in collections:
+            self.log.info("Extracted %d indicators for collection with name \"%s\"",
+                          coll["count"], coll["name"])
+
+        self.ioc_collections.extend(collections)
 
     def load_indicators_files(self, files, load_default=True):
         """
@@ -149,7 +186,8 @@ class Indicators:
             if os.path.isfile(file_path):
                 self.parse_stix2(file_path)
             else:
-                self.log.warning("This indicators file %s does not exist", file_path)
+                self.log.warning("No indicators file exists at path %s",
+                                 file_path)
 
         # Load downloaded indicators and any indicators from env variable.
         if load_default:
@@ -159,12 +197,13 @@ class Indicators:
         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, []):
+        for ioc_collection in self.ioc_collections:
+            for ioc in ioc_collection.get(ioc_type, []):
                 yield {
                     "value": ioc,
                     "type": ioc_type,
-                    "name": ioc_file["name"]
+                    "name": ioc_collection["name"],
+                    "stix2_file_name": ioc_collection["stix2_file_name"],
                 }
 
     def check_domain(self, url):
@@ -424,18 +463,18 @@ def download_indicators_files(log):
 
     for ioc_entry in res.json():
         ioc_url = ioc_entry["stix2_url"]
-        log.info("Downloading indicator file '%s' from '%s'", ioc_entry["name"], ioc_url)
+        log.info("Downloading indicator file %s from %s", ioc_entry["name"], ioc_url)
 
         res = requests.get(ioc_url)
         if res.status_code != 200:
-            log.warning("Could not find indicator file '%s'", ioc_url)
+            log.warning("Could not find indicator file %s", ioc_url)
             continue
 
         clean_file_name = ioc_url.lstrip("https://").replace("/", "_")
         ioc_path = os.path.join(data_dir, clean_file_name)
 
         # Write file to disk. This will overwrite any older version of the STIX2 file.
-        with open(ioc_path, "w", encoding="utf-8") as f:
-            f.write(res.text)
+        with open(ioc_path, "w", encoding="utf-8") as handle:
+            handle.write(res.text)
 
-        log.info("Saved indicator file to '%s'", os.path.basename(ioc_path))
+        log.info("Saved indicator file to %s", os.path.basename(ioc_path))

mvt/common/utils.py

@@ -73,7 +73,7 @@ def check_for_links(text):
     :returns: Search results.
 
     """
-    return re.findall("(?P<url>https?://[^\s]+)", text, re.IGNORECASE)
+    return re.findall(r"(?P<url>https?://[^\s]+)", text, re.IGNORECASE)
 
 
 def get_sha256_from_file_path(file_path):

mvt/common/version.py

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

mvt/ios/modules/backup/backup_info.py

@@ -7,6 +7,7 @@ import os
 import plistlib
 
 from mvt.common.module import DatabaseNotFoundError
+from mvt.ios.versions import latest_ios_version
 
 from ..base import IOSExtraction
 
@@ -41,3 +42,9 @@ class BackupInfo(IOSExtraction):
             value = info.get(field, None)
             self.log.info("%s: %s", field, value)
             self.results[field] = value
+
+        if "Product Version" in info:
+            latest = latest_ios_version()
+            if info["Product Version"] != latest["version"]:
+                self.log.warning("This phone is running an outdated iOS version: %s (latest is %s)",
+                                 info["Product Version"], latest['version'])

mvt/ios/versions.py

@@ -237,6 +237,8 @@ IPHONE_IOS_VERSIONS = [
     {"build": "19C56", "version": "15.2"},
     {"build": "19C63", "version": "15.2.1"},
     {"build": "19D50", "version": "15.3"},
+    {"build": "19D52", "version": "15.3.1"},
+    {"build": "19E241", "version": "15.4"}
 ]
 
 
@@ -251,3 +253,7 @@ def find_version_by_build(build):
     for version in IPHONE_IOS_VERSIONS:
         if build == version["build"]:
             return version["version"]
+
+
+def latest_ios_version():
+    return IPHONE_IOS_VERSIONS[-1]

setup.py

@@ -29,6 +29,7 @@ requires = (
     # Android dependencies:
     "adb-shell>=0.4.2",
     "libusb1>=2.0.1",
+    "cryptography>=36.0.1"
 )
 
 

tests/android/test_backup_module.py

@@ -0,0 +1,78 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2022 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
+import tarfile
+
+from mvt.android.modules.backup.sms import SMS
+from mvt.android.parsers.backup import parse_backup_file
+from mvt.common.module import run_module
+
+from ..utils import get_android_backup_folder
+
+
+class TestBackupModule:
+    def test_module_folder(self):
+        backup_path = get_android_backup_folder()
+        mod = SMS(base_folder=backup_path, log=logging)
+        files = []
+        for root, subdirs, subfiles in os.walk(os.path.abspath(backup_path)):
+            for fname in subfiles:
+                files.append(os.path.relpath(os.path.join(root, fname), backup_path))
+        mod.from_folder(backup_path, files)
+        run_module(mod)
+        assert len(mod.results) == 1
+        assert len(mod.results[0]["links"]) == 1
+        assert mod.results[0]["links"][0] == "https://google.com/"
+
+    def test_module_file(self):
+        fpath = os.path.join(get_android_backup_folder(), "backup.ab")
+        mod = SMS(base_folder=fpath, log=logging)
+        with open(fpath, "rb") as f:
+            data = f.read()
+        tardata = parse_backup_file(data)
+        dbytes = io.BytesIO(tardata)
+        tar = tarfile.open(fileobj=dbytes)
+        files = []
+        for member in tar:
+            files.append(member.name)
+        mod.from_ab(fpath, tar, files)
+        run_module(mod)
+        assert len(mod.results) == 1
+        assert len(mod.results[0]["links"]) == 1
+
+    def test_module_file2(self):
+        fpath = os.path.join(get_android_backup_folder(), "backup2.ab")
+        mod = SMS(base_folder=fpath, log=logging)
+        with open(fpath, "rb") as f:
+            data = f.read()
+        tardata = parse_backup_file(data, password="123456")
+        dbytes = io.BytesIO(tardata)
+        tar = tarfile.open(fileobj=dbytes)
+        files = []
+        for member in tar:
+            files.append(member.name)
+        mod.from_ab(fpath, tar, files)
+        run_module(mod)
+        assert len(mod.results) == 1
+        assert len(mod.results[0]["links"]) == 1
+
+    def test_module_file3(self):
+        fpath = os.path.join(get_android_backup_folder(), "backup3.ab")
+        mod = SMS(base_folder=fpath, log=logging)
+        with open(fpath, "rb") as f:
+            data = f.read()
+        tardata = parse_backup_file(data)
+        dbytes = io.BytesIO(tardata)
+        tar = tarfile.open(fileobj=dbytes)
+        files = []
+        for member in tar:
+            files.append(member.name)
+        mod.from_ab(fpath, tar, files)
+        run_module(mod)
+        assert len(mod.results) == 1
+        assert len(mod.results[0]["links"]) == 1

tests/android/test_backup_parser.py

@@ -0,0 +1,58 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2022 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 hashlib
+import logging
+
+from mvt.android.parsers.backup import parse_backup_file, parse_tar_for_sms
+
+from ..utils import get_artifact
+
+
+class TestBackupParsing:
+    def test_parsing_noencryption(self):
+        file = get_artifact("android_backup/backup.ab")
+        with open(file, "rb") as f:
+            data = f.read()
+        ddata = parse_backup_file(data)
+
+        m = hashlib.sha256()
+        m.update(ddata)
+        assert m.hexdigest() == "0799b583788908f06bccb854608cede375041ee878722703a39182edeb008324"
+        sms = parse_tar_for_sms(ddata)
+        assert isinstance(sms, list) == True
+        assert len(sms) == 1
+        assert len(sms[0]["links"]) == 1
+        assert sms[0]["links"][0] == "https://google.com/"
+
+    def test_parsing_encryption(self):
+        file = get_artifact("android_backup/backup2.ab")
+        with open(file, "rb") as f:
+            data = f.read()
+        ddata = parse_backup_file(data, password="123456")
+
+        m = hashlib.sha256()
+        m.update(ddata)
+        assert m.hexdigest() == "f365ace1effbc4902c6aeba241ca61544f8a96ad456c1861808ea87b7dd03896"
+        sms = parse_tar_for_sms(ddata)
+        assert isinstance(sms, list) == True
+        assert len(sms) == 1
+        assert len(sms[0]["links"]) == 1
+        assert sms[0]["links"][0] == "https://google.com/"
+
+    def test_parsing_compression(self):
+        file = get_artifact("android_backup/backup3.ab")
+        with open(file, "rb") as f:
+            data = f.read()
+        ddata = parse_backup_file(data)
+
+        m = hashlib.sha256()
+        m.update(ddata)
+        assert m.hexdigest() == "33e73df2ede9798dcb3a85c06200ee41c8f52dd2f2e50ffafcceb0407bc13e3a"
+        sms = parse_tar_for_sms(ddata)
+        assert isinstance(sms, list) == True
+        assert len(sms) == 1
+        assert len(sms[0]["links"]) == 1
+        assert sms[0]["links"][0] == "https://google.com/"

tests/artifacts/android_backup/apps/com.android.providers.telephony/d_f/000000_sms_backup

@@ -0,0 +1,2 @@
+xœEN]
+Â0¾Jè³Z«ku;<3B>wC꺂[ˆxwS„¼|ÿéßʇ°"‘êÔÑÙæ|2­Ú©G›0·<%9ŒÄ0‹ÕGÎ01ê´Ž9Ç'Æ<kIÏ(Iã+m—«i­ûÑwÂ…ëŽ`bϯ:º7‚x+5T<35>+Ž©$1ŠØÿ_ªâCmVŸáÖ5¡

tests/common/test_indicators.py

@@ -13,11 +13,11 @@ 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
+        assert ind.ioc_collections[0]["count"] == 4
+        assert len(ind.ioc_collections[0]["domains"]) == 1
+        assert len(ind.ioc_collections[0]["emails"]) == 1
+        assert len(ind.ioc_collections[0]["file_names"]) == 1
+        assert len(ind.ioc_collections[0]["processes"]) == 1
 
     def test_check_domain(self, indicator_file):
         ind = Indicators(log=logging)

tests/ios/test_backup_info.py

@@ -8,12 +8,12 @@ import logging
 from mvt.common.module import run_module
 from mvt.ios.modules.backup.backup_info import BackupInfo
 
-from ..utils import get_backup_folder
+from ..utils import get_ios_backup_folder
 
 
 class TestBackupInfoModule:
     def test_manifest(self):
-        m = BackupInfo(base_folder=get_backup_folder(), log=logging)
+        m = BackupInfo(base_folder=get_ios_backup_folder(), log=logging)
         run_module(m)
         assert m.results["Build Version"] == "18C66"
         assert m.results["IMEI"] == "42"

tests/ios/test_datausage.py

@@ -9,23 +9,23 @@ 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
+from ..utils import get_ios_backup_folder
 
 
 class TestDatausageModule:
     def test_datausage(self):
-        m = Datausage(base_folder=get_backup_folder(), log=logging, results=[])
+        m = Datausage(base_folder=get_ios_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=[])
+        m = Datausage(base_folder=get_ios_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")
+        ind.ioc_collections[0]["processes"].append("CumulativeUsageTracker")
         m.indicators = ind
         run_module(m)
         assert len(m.detected) == 2

tests/ios/test_manifest.py

@@ -9,22 +9,22 @@ 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
+from ..utils import get_ios_backup_folder
 
 
 class TestManifestModule:
     def test_manifest(self):
-        m = Manifest(base_folder=get_backup_folder(), log=logging, results=[])
+        m = Manifest(base_folder=get_ios_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=[])
+        m = Manifest(base_folder=get_ios_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")
+        ind.ioc_collections[0]["file_names"].append("com.apple.CoreBrightness.plist")
         m.indicators = ind
         run_module(m)
         assert len(m.detected) == 1

tests/ios/test_safari_browserstate.py

@@ -9,12 +9,12 @@ from mvt.common.indicators import Indicators
 from mvt.common.module import run_module
 from mvt.ios.modules.mixed.safari_browserstate import SafariBrowserState
 
-from ..utils import get_backup_folder
+from ..utils import get_ios_backup_folder
 
 
 class TestSafariBrowserStateModule:
     def test_parsing(self):
-        m = SafariBrowserState(base_folder=get_backup_folder(), log=logging, results=[])
+        m = SafariBrowserState(base_folder=get_ios_backup_folder(), log=logging, results=[])
         m.is_backup = True
         run_module(m)
         assert m.file_path is not None
@@ -23,12 +23,12 @@ class TestSafariBrowserStateModule:
         assert len(m.detected) == 0
 
     def test_detection(self, indicator_file):
-        m = SafariBrowserState(base_folder=get_backup_folder(), log=logging, results=[])
+        m = SafariBrowserState(base_folder=get_ios_backup_folder(), log=logging, results=[])
         m.is_backup = True
         ind = Indicators(log=logging)
         ind.parse_stix2(indicator_file)
         # Adds a file that exists in the manifest.
-        ind.ioc_files[0]["domains"].append("en.wikipedia.org")
+        ind.ioc_collections[0]["domains"].append("en.wikipedia.org")
         m.indicators = ind
         run_module(m)
         assert len(m.detected) == 1

tests/ios/test_sms.py

@@ -0,0 +1,31 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2022 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.sms import SMS
+
+from ..utils import get_ios_backup_folder
+
+
+class TestSMSModule:
+    def test_sms(self):
+        m = SMS(base_folder=get_ios_backup_folder(), log=logging, results=[])
+        run_module(m)
+        assert len(m.results) == 1
+        assert len(m.timeline) == 1
+        assert len(m.detected) == 0
+
+    def test_detection(self, indicator_file):
+        m = SMS(base_folder=get_ios_backup_folder(), log=logging, results=[])
+        ind = Indicators(log=logging)
+        ind.parse_stix2(indicator_file)
+        # Adds a file that exists in the manifest.
+        ind.ioc_collections[0]["domains"].append("badbadbad.example.org")
+        m.indicators = ind
+        run_module(m)
+        assert len(m.detected) == 1

tests/ios/test_tcc.py

@@ -9,12 +9,12 @@ 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
+from ..utils import get_ios_backup_folder
 
 
 class TestTCCtModule:
     def test_tcc(self):
-        m = TCC(base_folder=get_backup_folder(), log=logging, results=[])
+        m = TCC(base_folder=get_ios_backup_folder(), log=logging, results=[])
         run_module(m)
         assert len(m.results) == 11
         assert len(m.timeline) == 11
@@ -24,7 +24,7 @@ class TestTCCtModule:
         assert m.results[0]["auth_value"] == "allowed"
 
     def test_tcc_detection(self, indicator_file):
-        m = TCC(base_folder=get_backup_folder(), log=logging, results=[])
+        m = TCC(base_folder=get_ios_backup_folder(), log=logging, results=[])
         ind = Indicators(log=logging)
         ind.parse_stix2(indicator_file)
         m.indicators = ind

tests/test_check_backup.py

@@ -0,0 +1,18 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2022 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/
+
+from click.testing import CliRunner
+
+from mvt.ios.cli import check_backup
+
+from .utils import get_ios_backup_folder
+
+
+class TestCheckBackupCommand:
+    def test_check(self):
+        runner = CliRunner()
+        path = get_ios_backup_folder()
+        result = runner.invoke(check_backup, [path])
+        assert result.exit_code == 0

tests/utils.py

@@ -20,9 +20,13 @@ def get_artifact_folder():
     return os.path.join(os.path.dirname(__file__), "artifacts")
 
 
-def get_backup_folder():
+def get_ios_backup_folder():
     return os.path.join(os.path.dirname(__file__), "artifacts", "ios_backup")
 
 
+def get_android_backup_folder():
+    return os.path.join(os.path.dirname(__file__), "artifacts", "android_backup")
+
+
 def get_indicator_file():
     print("PYTEST env", os.getenv("PYTEST_CURRENT_TEST"))