Bumped version

Add initial implementation of SMS extraction using ADB

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,14 +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)
@@ -51,7 +57,7 @@ def version():
 #==============================================================================
 # 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")
@@ -246,6 +252,44 @@ def check_bugreport(ctx, iocs, output, list_modules, module, bugreport_path):
 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)
@@ -256,14 +300,6 @@ 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.basename(backup_path) == "backup.ab":
-            log.info("You can use ABE (https://github.com/nelenkov/android-backup-extractor) "
-                     "to extract 'backup.ab' files!")
-        ctx.exit(1)
-
     for module in BACKUP_MODULES:
         m = module(base_folder=backup_path, output_folder=output,
                    log=logging.getLogger(module.__module__))
@@ -273,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/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/getprop.py

@@ -5,6 +5,7 @@
 
 import logging
 import re
+from datetime import datetime, timedelta
 
 from mvt.android.parsers import parse_getprop
 
@@ -31,4 +32,12 @@ class Getprop(AndroidExtraction):
 
         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/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/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/getprop.py

@@ -5,6 +5,7 @@
 
 import logging
 import re
+from datetime import datetime, timedelta
 
 from mvt.android.parsers import parse_getprop
 
@@ -47,4 +48,12 @@ class Getprop(BugReportModule):
 
         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/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/common/version.py

@@ -6,7 +6,7 @@
 import requests
 from packaging import version
 
-MVT_VERSION = "1.5.1"
+MVT_VERSION = "1.5.2"
 
 
 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,7 @@ 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"}
 ]
 
 
@@ -251,3 +252,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/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,19 +9,19 @@ 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.

tests/ios/test_manifest.py

@@ -9,19 +9,19 @@ 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_collections[0]["file_names"].append("com.apple.CoreBrightness.plist")

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,7 +23,7 @@ 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)

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