Merge pull request #251 from mvt-project/feature/read-sms-adb-backup

Add initial implementation of SMS extraction using ADB

mvt/android/cli.py

@@ -5,6 +5,9 @@
 
 import logging
 import os
+import getpass
+import io
+import tarfile
 from pathlib import Path
 from zipfile import ZipFile
 
@@ -17,6 +20,8 @@ from mvt.common.help import (HELP_MSG_FAST, HELP_MSG_IOC,
 from mvt.common.indicators import Indicators, download_indicators_files
 from mvt.common.logo import logo
 from mvt.common.module import run_module, save_timeline
+from mvt.android.parsers.backup import parse_ab_header, parse_backup_file
+from mvt.android.parsers.backup import InvalidBackupPassword, AndroidBackupParsingError
 
 from .download_apks import DownloadAPKs
 from .lookups.koodous import koodous_lookup
@@ -246,6 +251,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 +299,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 +308,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

@@ -10,6 +10,8 @@ import string
 import sys
 import tempfile
 import time
+import base64
+import getpass
 
 from adb_shell.adb_device import AdbDeviceTcp, AdbDeviceUsb
 from adb_shell.auth.keygen import keygen, write_public_keyfile
@@ -19,6 +21,7 @@ from adb_shell.exceptions import (AdbCommandFailureException, DeviceAuthError,
 from usb1 import USBErrorAccess, USBErrorBusy
 
 from mvt.common.module import InsufficientPrivileges, MVTModule
+from mvt.android.parsers.backup import parse_ab_header, parse_backup_file, InvalidBackupPassword
 
 log = logging.getLogger(__name__)
 
@@ -243,6 +246,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/sms.py

@@ -6,8 +6,12 @@
 import logging
 import os
 import sqlite3
+import base64
+import getpass
 
 from mvt.common.utils import check_for_links, convert_timestamp_to_iso
+from mvt.android.parsers.backup import parse_tar_for_sms, AndroidBackupParsingError
+from mvt.common.module import InsufficientPrivileges
 
 from .base import AndroidExtraction
 
@@ -16,11 +20,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 +36,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 +54,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 +67,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 +84,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 +101,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 +109,36 @@ 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 os
+import fnmatch
+
+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.common.utils import check_for_links
+from mvt.android.parsers.backup import parse_sms_file
 
 
-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/parsers/backup.py

@@ -0,0 +1,199 @@
+# 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 zlib
+import json
+import tarfile
+import datetime
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
+from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
+from cryptography.hazmat.primitives import padding
+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

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 logging
+import os
+import tarfile
+import io
+
+from mvt.android.modules.backup.sms import SMS
+from mvt.common.module import run_module
+from mvt.android.parsers.backup import parse_backup_file
+
+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 logging
+import hashlib
+
+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;�wC꺂[ˆxwS„¼|ÿéßʇ°"‘êÔÑÙæ|2­Ú©G›0·<%9ŒÄ0‹ÕGÎ01ê´Ž9Ç'Æ<kIÏ(Iã+m—«i­ûÑwÂ…ëŽ`bϯ:º7‚x+5T�+Ž©$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

@@ -9,19 +9,19 @@ 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_backup_folder
+from ..utils import get_ios_backup_folder
 
 
 class TestSMSModule:
     def test_sms(self):
-        m = SMS(base_folder=get_backup_folder(), log=logging, results=[])
+        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_backup_folder(), log=logging, results=[])
+        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.

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

@@ -7,12 +7,12 @@ from click.testing import CliRunner
 
 from mvt.ios.cli import check_backup
 
-from .utils import get_backup_folder
+from .utils import get_ios_backup_folder
 
 
 class TestCheckBackupCommand:
     def test_check(self):
         runner = CliRunner()
-        path = get_backup_folder()
+        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"))