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/`.

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))