diff --git a/mvt/android/cli.py b/mvt/android/cli.py index e070edb..9e7d220 100644 --- a/mvt/android/cli.py +++ b/mvt/android/cli.py @@ -256,14 +256,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__)) diff --git a/mvt/android/modules/adb/sms.py b/mvt/android/modules/adb/sms.py index 0798f50..b8f9333 100644 --- a/mvt/android/modules/adb/sms.py +++ b/mvt/android/modules/adb/sms.py @@ -6,14 +6,11 @@ import logging import os import sqlite3 -import tarfile -import io -import zlib import base64 -import json -import datetime +import getpass from mvt.common.utils import check_for_links, convert_timestamp_to_iso +from mvt.android.parsers.backup import parse_ab_header, parse_sms_backup, InvalidBackupPassword from mvt.common.module import InsufficientPrivileges from .base import AndroidExtraction @@ -73,6 +70,7 @@ class SMS(AndroidExtraction): if "body" not in message: continue + # 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) @@ -86,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] @@ -111,29 +109,6 @@ class SMS(AndroidExtraction): log.info("Extracted a total of %d SMS messages containing links", len(self.results)) - def _extract_sms_from_backup_tar(self, tar_data): - # Extract data from generated tar file - tar_bytes = io.BytesIO(tar_data) - tar = tarfile.open(fileobj=tar_bytes, mode='r') - for member in tar.getmembers(): - if not member.name.endswith("_sms_backup"): - continue - - self.log.debug("Extracting SMS messages from backup file %s", member.name) - sms_part_zlib = zlib.decompress(tar.extractfile(member).read()) - json_data = json.loads(sms_part_zlib) - - # TODO: Copied from SMS module. Refactor to avoid duplication - for message in json_data: - utc_timestamp = datetime.datetime.utcfromtimestamp(int(message["date"]) / 1000) - message["isodate"] = convert_timestamp_to_iso(utc_timestamp) - message["direction"] = ("sent" if int(message["date_sent"]) else "received") - - message_links = check_for_links(message["body"]) - if message_links or message["body"].strip() == "": - self.results.append(message) - - 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 @@ -141,23 +116,33 @@ class SMS(AndroidExtraction): 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. """ + # Run ADB command to create a backup of SMS app self.log.warning("Please check phone and accept Android backup prompt. Do not set an encryption 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 com.android.providers.telephony | base64") backup_output = base64.b64decode(backup_output_b64) - if not backup_output.startswith(b"ANDROID BACKUP"): + 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 - [magic_header, version, is_compressed, encryption, tar_data] = backup_output.split(b"\n", 4) - if encryption != b"none" or int(is_compressed): - self.log.error("The backup is encrypted or compressed and cannot be parsed. " - "[version: %s, encryption: %s, compression: %s]", version, encryption, is_compressed) + if header["compression"]: + self.log.error("The backup is compressed and cannot be parsed, quitting...") return - self._extract_sms_from_backup_tar(tar_data) + pwd = None + if header["encryption"] != "none": + pwd = getpass.getpass(prompt="Backup Password: ", stream=None) + + + try: + self.results = parse_sms_backup(backup_output, password=pwd) + except InvalidBackupPassword: + self.info.log("Invalid backup password") + return + log.info("Extracted a total of %d SMS messages containing links", len(self.results)) def run(self): try: diff --git a/mvt/android/modules/backup/sms.py b/mvt/android/modules/backup/sms.py index 4ab2b1f..12d3384 100644 --- a/mvt/android/modules/backup/sms.py +++ b/mvt/android/modules/backup/sms.py @@ -3,16 +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 json import os -import zlib +import getpass from mvt.common.module import MVTModule from mvt.common.utils import check_for_links +from mvt.android.parsers.backup import parse_sms_file, parse_sms_backup, parse_ab_header, InvalidBackupPassword class SMS(MVTModule): - 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, @@ -35,30 +34,49 @@ class SMS(MVTModule): 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) + data = handle.read() - 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) + self.results = parse_sms_file(data) 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") + # FIXME: this should be done in the Module code if there are other modules on backups + if os.path.isfile(self.base_folder): + # ab file + with open(self.base_folder, "rb") as handle: + data = handle.read() + header = parse_ab_header(data) + if not header["backup"]: + self.log.info("Not a valid Android Backup file, quitting...") + return + if header["compression"]: + self.log.info("MVT does not support compressed backups, either regenerate the backup with the -nocompress option or use ANdroid Backup Extractor to convert it to a tar file") + self.log.info("Quitting...") + return + pwd = None + if header["encryption"] != "none": + pwd = getpass.getpass(prompt="Backup Password: ", stream=None) - for file_name in os.listdir(app_folder): - if not file_name.endswith("_sms_backup"): - continue + try: + messages = parse_sms_backup(data, password=pwd) + except InvalidBackupPassword: + self.log.info("Invalid password, impossible de decrypt the backup, quitting...") + return + self.results = messages + else: + app_folder = os.path.join(self.base_folder, + "apps", + "com.android.providers.telephony", + "d_f") + if not os.path.exists(app_folder): + self.log.info("Unable to find the SMS backup folder") + return - file_path = os.path.join(app_folder, file_name) - self._process_sms_file(file_path) + 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) self.log.info("Extracted a total of %d SMS messages containing links", len(self.results)) diff --git a/mvt/android/parsers/backup.py b/mvt/android/parsers/backup.py new file mode 100644 index 0000000..9a83b84 --- /dev/null +++ b/mvt/android/parsers/backup.py @@ -0,0 +1,195 @@ +# 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 hashlib +import datetime +import pyaes +from mvt.common.utils import check_for_links, convert_timestamp_to_iso + + +PBKDF2_KEY_SIZE = 32 + + +class AndroidBackupParseError(Exception): + """Exception raised file parsing an android backup file""" + + +class AndroidBackupNotImplemented(AndroidBackupParseError): + pass + + +class InvalidBackupPassword(AndroidBackupParseError): + pass + + +def decrypt_master_key_blob(key, aes_iv, cipher_text): + """ + Decrypt the master key blob with AES + From : https://github.com/FloatingOctothorpe/dump_android_backup + """ + + aes = pyaes.AESModeOfOperationCBC(key, aes_iv) + + plain_text = b'' + while len(plain_text) < len(cipher_text): + offset = len(plain_text) + plain_text += aes.decrypt(cipher_text[offset:(offset + 16)]) + + blob = io.BytesIO(plain_text) + master_iv_length = ord(blob.read(1)) + master_iv = blob.read(master_iv_length) + master_key_length = ord(blob.read(1)) + master_key = blob.read(master_key_length) + master_key_checksum_length = ord(blob.read(1)) + master_key_checksum = blob.read(master_key_checksum_length) + + return master_iv, master_key, master_key_checksum + + +def to_utf8_bytes(input_bytes): + """Emulate bytes being converted into a "UTF8 byte array" + For more info see the Bouncy Castle Crypto package Strings.toUTF8ByteArray + method: + https://github.com/bcgit/bc-java/blob/master/core/src/main/java/org/bouncycastle/util/Strings.java#L142 + From https://github.com/FloatingOctothorpe/dump_android_backup + """ + 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_sms_backup(data, password=None): + """ + Parse a backup file and returns SMS in it + """ + tardata = parse_backup_file(data, password=password) + return parse_tar_for_sms(tardata) + + +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 parse_backup_file(data, password=None): + """ + Parse an ab file, returns a tar file + Inspired by https://github.com/FloatingOctothorpe/dump_android_backup + """ + if not data.startswith(b"ANDROID BACKUP"): + raise AndroidBackupParseError("Invalid file header") + + [magic_header, version, is_compressed, encryption, tar_data] = data.split(b"\n", 4) + version = int(version) + is_compressed = int(is_compressed) + if is_compressed == 1: + raise AndroidBackupNotImplemented("Compression is not implemented") + + if encryption != b"none": + if encryption != 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] = tar_data.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")) + + key = hashlib.pbkdf2_hmac('sha1', + password.encode('utf-8'), + user_salt, + pbkdf2_rounds, + PBKDF2_KEY_SIZE) + try: + [master_iv, master_key, master_key_checksum] = decrypt_master_key_blob(key, user_iv, master_key_blob) + except TypeError: + raise InvalidBackupPassword() + + if version > 1: + hmac_mk = to_utf8_bytes(master_key) + else: + hmac_mk = master_key + + calculated_checksum = hashlib.pbkdf2_hmac('sha1', + hmac_mk, + checksum_salt, + pbkdf2_rounds, + PBKDF2_KEY_SIZE) + + if master_key_checksum != calculated_checksum: + raise InvalidBackupPassword() + + decrypter = pyaes.Decrypter(pyaes.AESModeOfOperationCBC(master_key, master_iv)) + tar_data = decrypter.feed(encrypted_data) + + 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 diff --git a/setup.py b/setup.py index fe71b9e..ea00db6 100755 --- a/setup.py +++ b/setup.py @@ -29,6 +29,7 @@ requires = ( # Android dependencies: "adb-shell>=0.4.2", "libusb1>=2.0.1", + "pyaes>=1.6.1" ) diff --git a/tests/android/__init__.py b/tests/android/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/android/test_backup_module.py b/tests/android/test_backup_module.py new file mode 100644 index 0000000..347e4e3 --- /dev/null +++ b/tests/android/test_backup_module.py @@ -0,0 +1,29 @@ +# 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 + +from mvt.android.modules.backup.sms import SMS +from mvt.common.module import run_module + +from ..utils import get_android_backup_folder + + +class TestBackupModule: + def test_module_folder(self): + fpath = get_android_backup_folder() + mod = SMS(base_folder=fpath, log=logging) + 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) + run_module(mod) + assert len(mod.results) == 1 + assert len(mod.results[0]["links"]) == 1 diff --git a/tests/android/test_backup_parser.py b/tests/android/test_backup_parser.py new file mode 100644 index 0000000..ca754cb --- /dev/null +++ b/tests/android/test_backup_parser.py @@ -0,0 +1,50 @@ +# 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/" + + + + + + + diff --git a/tests/artifacts/android_backup/apps/com.android.providers.telephony/d_f/000000_sms_backup b/tests/artifacts/android_backup/apps/com.android.providers.telephony/d_f/000000_sms_backup new file mode 100644 index 0000000..14d85a2 --- /dev/null +++ b/tests/artifacts/android_backup/apps/com.android.providers.telephony/d_f/000000_sms_backup @@ -0,0 +1,2 @@ +xEN] +0 JZku;wC[xwS|ʇ"|2کG0<%90G01괎9'