Adds partial compression support in Android Backup parsing

mvt/android/modules/adb/sms.py

@@ -10,7 +10,7 @@ import base64
 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.android.parsers.backup import parse_ab_header, parse_sms_backup, InvalidBackupPassword, AndroidBackupParsingError
 from mvt.common.module import InsufficientPrivileges
 
 from .base import AndroidExtraction
@@ -128,10 +128,6 @@ class SMS(AndroidExtraction):
             self.log.error("Extracting SMS via Android backup failed. No valid backup data found.")
             return
 
-        if header["compression"]:
-            self.log.error("The backup is compressed and cannot be parsed, quitting...")
-            return
-
         pwd = None
         if header["encryption"] != "none":
             pwd = getpass.getpass(prompt="Backup Password: ", stream=None)
@@ -140,8 +136,12 @@ class SMS(AndroidExtraction):
         try:
             self.results = parse_sms_backup(backup_output, password=pwd)
         except InvalidBackupPassword:
-            self.info.log("Invalid backup password")
+            self.log.info("Invalid backup password")
             return
+        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):

mvt/android/modules/backup/sms.py

@@ -8,7 +8,7 @@ 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
+from mvt.android.parsers.backup import parse_sms_file, parse_sms_backup, parse_ab_header, InvalidBackupPassword, AndroidBackupParsingError
 
 
 class SMS(MVTModule):
@@ -48,10 +48,7 @@ class SMS(MVTModule):
             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)
@@ -61,6 +58,11 @@ class SMS(MVTModule):
             except InvalidBackupPassword:
                 self.log.info("Invalid password, impossible de decrypt the backup, quitting...")
                 return
+            except AndroidBackupParsingError:
+                self.log.info("Impossible to extract data from this Android Backup, please regenerate the backup using the -nocompress option or extract it using Android Backup Extractor instead.")
+                self.log.info("Quitting...")
+                return
+
             self.results = messages
         else:
             app_folder = os.path.join(self.base_folder,

mvt/android/parsers/backup.py

@@ -16,15 +16,15 @@ from mvt.common.utils import check_for_links, convert_timestamp_to_iso
 PBKDF2_KEY_SIZE = 32
 
 
-class AndroidBackupParseError(Exception):
+class AndroidBackupParsingError(Exception):
     """Exception raised file parsing an android backup file"""
 
 
-class AndroidBackupNotImplemented(AndroidBackupParseError):
+class AndroidBackupNotImplemented(AndroidBackupParsingError):
     pass
 
 
-class InvalidBackupPassword(AndroidBackupParseError):
+class InvalidBackupPassword(AndroidBackupParsingError):
     pass
 
 
@@ -106,13 +106,11 @@ def parse_backup_file(data, password=None):
     Inspired by https://github.com/FloatingOctothorpe/dump_android_backup
     """
     if not data.startswith(b"ANDROID BACKUP"):
-        raise AndroidBackupParseError("Invalid file header")
+        raise AndroidBackupParsingError("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":
@@ -153,6 +151,12 @@ def parse_backup_file(data, password=None):
         decrypter = pyaes.Decrypter(pyaes.AESModeOfOperationCBC(master_key, master_iv))
         tar_data = decrypter.feed(encrypted_data)
 
+    if is_compressed == 1:
+        try:
+            tar_data = zlib.decompress(tar_data)
+        except zlib.error:
+            raise AndroidBackupParsingError("Impossible to decompress the file")
+
     return tar_data
 
 

tests/android/test_backup_parser.py

@@ -42,9 +42,17 @@ class TestBackupParsing:
         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/"