Merge pull request #370 from mvt-project/android-backup-refactor

Refactor Android backup password handling and add tests

docs/android/backup.md

@@ -35,7 +35,11 @@ $ mvt-android check-backup --output /path/to/results/ /path/to/backup.ab
          INFO     [mvt.android.modules.backup.sms] Extracted a total of 64 SMS messages
 ```
 
-If the backup is encrypted, MVT will prompt you to enter the password.
+If the backup is encrypted, MVT will prompt you to enter the password. A backup password can also be provided with the `--backup-password` command line option or through the `MVT_ANDROID_BACKUP_PASSWORD` environment variable. The same options can also be used to when analysing an encrypted backup collected through AndroidQF in the `mvt-android check-androidqf` command:
+
+```bash
+$ mvt-android check-backup --backup-password "password123" --output /path/to/results/ /path/to/backup.ab
+```
 
 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.
 

mvt/android/cli.py

@@ -4,7 +4,6 @@
 #   https://license.mvt.re/1.1/
 
 import logging
-import os
 
 import click
 
@@ -34,11 +33,11 @@ from .modules.adb import ADB_MODULES
 from .modules.adb.packages import Packages
 from .modules.backup import BACKUP_MODULES
 from .modules.bugreport import BUGREPORT_MODULES
+from .modules.backup.helpers import cli_load_android_backup_password
 
 init_logging()
 log = logging.getLogger("mvt")
 
-MVT_ANDROID_BACKUP_PASSWORD = "MVT_ANDROID_BACKUP_PASSWORD"
 CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])
 
 
@@ -130,7 +129,7 @@ def download_apks(ctx, all_apks, virustotal, output, from_file, serial, verbose)
 # ==============================================================================
 @cli.command(
     "check-adb",
-    help="Check an Android device over adb",
+    help="Check an Android device over ADB",
     context_settings=CONTEXT_SETTINGS,
 )
 @click.option("--serial", "-s", type=str, help=HELP_MSG_SERIAL)
@@ -146,11 +145,28 @@ def download_apks(ctx, all_apks, virustotal, output, from_file, serial, verbose)
 @click.option("--fast", "-f", is_flag=True, help=HELP_MSG_FAST)
 @click.option("--list-modules", "-l", is_flag=True, help=HELP_MSG_LIST_MODULES)
 @click.option("--module", "-m", help=HELP_MSG_MODULE)
+@click.option("--non-interactive", "-n", is_flag=True, help=HELP_MSG_NONINTERACTIVE)
+@click.option("--backup-password", "-p", help=HELP_MSG_ANDROID_BACKUP_PASSWORD)
 @click.option("--verbose", "-v", is_flag=True, help=HELP_MSG_VERBOSE)
 @click.pass_context
-def check_adb(ctx, serial, iocs, output, fast, list_modules, module, verbose):
+def check_adb(
+    ctx,
+    serial,
+    iocs,
+    output,
+    fast,
+    list_modules,
+    module,
+    non_interactive,
+    backup_password,
+    verbose,
+):
     set_verbose_logging(verbose)
-    module_options = {"fast_mode": fast}
+    module_options = {
+        "fast_mode": fast,
+        "interactive": not non_interactive,
+        "backup_password": cli_load_android_backup_password(log, backup_password),
+    }
 
     cmd = CmdAndroidCheckADB(
         results_path=output,
@@ -256,24 +272,6 @@ def check_backup(
 ):
     set_verbose_logging(verbose)
 
-    if backup_password:
-        log.info(
-            "Your password may be visible in the process table because it "
-            "was supplied on the command line!"
-        )
-
-        if MVT_ANDROID_BACKUP_PASSWORD in os.environ:
-            log.info(
-                "Ignoring %s environment variable, using --backup-password argument instead",
-                MVT_ANDROID_BACKUP_PASSWORD,
-            )
-    elif MVT_ANDROID_BACKUP_PASSWORD in os.environ:
-        log.info(
-            "Using backup password from %s environment variable",
-            MVT_ANDROID_BACKUP_PASSWORD,
-        )
-        backup_password = os.environ[MVT_ANDROID_BACKUP_PASSWORD]
-
     # Always generate hashes as backups are generally small.
     cmd = CmdAndroidCheckBackup(
         target_path=backup_path,
@@ -282,7 +280,7 @@ def check_backup(
         hashes=True,
         module_options={
             "interactive": not non_interactive,
-            "backup_password": backup_password,
+            "backup_password": cli_load_android_backup_password(log, backup_password),
         },
     )
 
@@ -340,24 +338,6 @@ def check_androidqf(
 ):
     set_verbose_logging(verbose)
 
-    if backup_password:
-        log.info(
-            "Your password may be visible in the process table because it "
-            "was supplied on the command line!"
-        )
-
-        if MVT_ANDROID_BACKUP_PASSWORD in os.environ:
-            log.info(
-                "Ignoring %s environment variable, using --backup-password argument instead",
-                MVT_ANDROID_BACKUP_PASSWORD,
-            )
-    elif MVT_ANDROID_BACKUP_PASSWORD in os.environ:
-        log.info(
-            "Using backup password from %s environment variable",
-            MVT_ANDROID_BACKUP_PASSWORD,
-        )
-        backup_password = os.environ[MVT_ANDROID_BACKUP_PASSWORD]
-
     cmd = CmdAndroidCheckAndroidQF(
         target_path=androidqf_path,
         results_path=output,
@@ -366,7 +346,7 @@ def check_androidqf(
         hashes=hashes,
         module_options={
             "interactive": not non_interactive,
-            "backup_password": backup_password,
+            "backup_password": cli_load_android_backup_password(log, backup_password),
         },
     )
 

mvt/android/cmd_check_backup.py

@@ -11,8 +11,6 @@ import tarfile
 from pathlib import Path
 from typing import List, Optional
 
-from rich.prompt import Prompt
-
 from mvt.android.modules.backup.base import BackupExtraction
 from mvt.android.parsers.backup import (
     AndroidBackupParsingError,
@@ -20,6 +18,7 @@ from mvt.android.parsers.backup import (
     parse_ab_header,
     parse_backup_file,
 )
+from mvt.android.modules.backup.helpers import prompt_or_load_android_backup_password
 from mvt.common.command import Command
 
 from .modules.backup import BACKUP_MODULES
@@ -72,7 +71,12 @@ class CmdAndroidCheckBackup(Command):
 
             password = None
             if header["encryption"] != "none":
-                password = Prompt.ask("Enter backup password", password=True)
+                password = prompt_or_load_android_backup_password(
+                    log, self.module_options
+                )
+                if not password:
+                    log.critical("No backup password provided.")
+                    sys.exit(1)
             try:
                 tardata = parse_backup_file(data, password=password)
             except InvalidBackupPassword:

mvt/android/modules/adb/base.py

@@ -22,7 +22,6 @@ from adb_shell.exceptions import (
     UsbDeviceNotFoundError,
     UsbReadFailedError,
 )
-from rich.prompt import Prompt
 from usb1 import USBErrorAccess, USBErrorBusy
 
 from mvt.android.parsers.backup import (
@@ -30,6 +29,7 @@ from mvt.android.parsers.backup import (
     parse_ab_header,
     parse_backup_file,
 )
+from mvt.android.modules.backup.helpers import prompt_or_load_android_backup_password
 from mvt.common.module import InsufficientPrivileges, MVTModule
 
 ADB_KEY_PATH = os.path.expanduser("~/.android/adbkey")
@@ -311,6 +311,12 @@ class AndroidExtraction(MVTModule):
             "You may need to set a backup password. \a"
         )
 
+        if self.module_options.get("backup_password", None):
+            self.log.warning(
+                "Backup password already set from command line or environment "
+                "variable. You should use the same password if enabling encryption!"
+            )
+
         # TODO: Base64 encoding as temporary fix to avoid byte-mangling over
         #       the shell transport...
         cmd = f"/system/bin/bu backup -nocompress '{package_name}' | base64"
@@ -329,7 +335,12 @@ class AndroidExtraction(MVTModule):
             return parse_backup_file(backup_output, password=None)
 
         for _ in range(0, 3):
-            backup_password = Prompt.ask("Enter backup password", password=True)
+            backup_password = prompt_or_load_android_backup_password(
+                self.log, self.module_options
+            )
+            if not backup_password:
+                # Fail as no backup password loaded for this encrypted backup
+                self.log.critical("No backup password provided.")
             try:
                 decrypted_backup_tar = parse_backup_file(backup_output, backup_password)
                 return decrypted_backup_tar

mvt/android/modules/androidqf/sms.py

@@ -4,7 +4,6 @@
 #   https://license.mvt.re/1.1
 
 import logging
-from rich.prompt import Prompt
 from typing import Optional
 
 from mvt.android.parsers.backup import (
@@ -14,6 +13,7 @@ from mvt.android.parsers.backup import (
     parse_backup_file,
     parse_tar_for_sms,
 )
+from mvt.android.modules.backup.helpers import prompt_or_load_android_backup_password
 
 from .base import AndroidQFModule
 
@@ -56,28 +56,19 @@ class SMS(AndroidQFModule):
             self.log.critical("Invalid backup format, backup.ab was not analysed")
             return
 
-        # the default is to allow interactivity
-        interactive = (
-            "interactive" not in self.module_options
-            or self.module_options["interactive"]
-        )
         password = None
         if header["encryption"] != "none":
-            if "backup_password" in self.module_options:
+            password = prompt_or_load_android_backup_password(
-                password = self.module_options["backup_password"]
+                self.log, self.module_options
-            elif interactive:
+            )
-                password = Prompt.ask(prompt="Enter backup password", password=True)
+            if not password:
-            else:
+                self.log.critical("No backup password provided.")
-                self.log.warning(
+                return
-                    "Cannot decrypt backup, because interactivity"
+
-                    " was disabled and the password was not"
-                    " supplied"
-                )
         try:
             tardata = parse_backup_file(data, password=password)
         except InvalidBackupPassword:
-            if "backup_password" in self.module_options or interactive:
+            self.log.critical("Invalid backup password")
-                self.log.critical("Invalid backup password")
             return
         except AndroidBackupParsingError:
             self.log.critical(

mvt/android/modules/backup/helpers.py

@@ -0,0 +1,61 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2023 Claudio Guarnieri.
+# 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
+
+from rich.prompt import Prompt
+
+
+MVT_ANDROID_BACKUP_PASSWORD = "MVT_ANDROID_BACKUP_PASSWORD"
+
+
+def cli_load_android_backup_password(log, backup_password):
+    """
+    Helper to load a backup password from CLI argument or environment variable
+
+    Used in MVT CLI command parsers.
+    """
+    password_from_env = os.environ.get(MVT_ANDROID_BACKUP_PASSWORD, None)
+    if backup_password:
+        log.info(
+            "Your password may be visible in the process table because it "
+            "was supplied on the command line!"
+        )
+        if password_from_env:
+            log.info(
+                "Ignoring %s environment variable, using --backup-password argument instead",
+                MVT_ANDROID_BACKUP_PASSWORD,
+            )
+        return backup_password
+    elif password_from_env:
+        log.info(
+            "Using backup password from %s environment variable",
+            MVT_ANDROID_BACKUP_PASSWORD,
+        )
+        return password_from_env
+
+
+def prompt_or_load_android_backup_password(log, module_options):
+    """
+    Used in modules to either prompt or load backup password to use for encryption and decryption.
+    """
+    if module_options.get("backup_password", None):
+        backup_password = module_options["backup_password"]
+        log.info(
+            "Using backup password passed from command line or environment variable."
+        )
+
+    # The default is to allow interactivity
+    elif module_options.get("interactive", True):
+        backup_password = Prompt.ask(prompt="Enter backup password", password=True)
+    else:
+        log.critical(
+            "Cannot decrypt backup because interactivity"
+            " was disabled and the password was not"
+            " supplied"
+        )
+        return None
+
+    return backup_password

tests/android_androidqf/test_sms.py

@@ -11,6 +11,8 @@ from mvt.common.module import run_module
 
 from ..utils import get_artifact_folder
 
+TEST_BACKUP_PASSWORD = "123456"
+
 
 class TestAndroidqfSMSAnalysis:
     def test_androidqf_sms(self):
@@ -21,3 +23,50 @@ class TestAndroidqfSMSAnalysis:
         assert len(m.results) == 2
         assert len(m.timeline) == 0
         assert len(m.detected) == 0
+
+    def test_androidqf_sms_encrypted_password_valid(self):
+        m = SMS(
+            target_path=os.path.join(get_artifact_folder(), "androidqf_encrypted"),
+            log=logging,
+            module_options={"backup_password": TEST_BACKUP_PASSWORD},
+        )
+        run_module(m)
+        assert len(m.results) == 1
+
+    def test_androidqf_sms_encrypted_password_prompt(self, mocker):
+        prompt_mock = mocker.patch(
+            "rich.prompt.Prompt.ask", return_value=TEST_BACKUP_PASSWORD
+        )
+        m = SMS(
+            target_path=os.path.join(get_artifact_folder(), "androidqf_encrypted"),
+            log=logging,
+            module_options={},
+        )
+        run_module(m)
+        assert prompt_mock.call_count == 1
+        assert len(m.results) == 1
+
+    def test_androidqf_sms_encrypted_password_invalid(self, caplog):
+        with caplog.at_level(logging.CRITICAL):
+            m = SMS(
+                target_path=os.path.join(get_artifact_folder(), "androidqf_encrypted"),
+                log=logging,
+                module_options={"backup_password": "invalid_password"},
+            )
+            run_module(m)
+        assert len(m.results) == 0
+        assert "Invalid backup password" in caplog.text
+
+    def test_androidqf_sms_encrypted_no_interactive(self, caplog):
+        with caplog.at_level(logging.CRITICAL):
+            m = SMS(
+                target_path=os.path.join(get_artifact_folder(), "androidqf_encrypted"),
+                log=logging,
+                module_options={"interactive": False},
+            )
+            run_module(m)
+        assert len(m.results) == 0
+        assert (
+            "Cannot decrypt backup because interactivity was disabled and the password was not supplied"
+            in caplog.text
+        )

tests/test_check_android_androidqf.py

@@ -12,9 +12,55 @@ from mvt.android.cli import check_androidqf
 from .utils import get_artifact_folder
 
 
+TEST_BACKUP_PASSWORD = "123456"
+
+
 class TestCheckAndroidqfCommand:
     def test_check(self):
         runner = CliRunner()
         path = os.path.join(get_artifact_folder(), "androidqf")
         result = runner.invoke(check_androidqf, [path])
         assert result.exit_code == 0
+
+    def test_check_encrypted_backup_prompt_valid(self, mocker):
+        """Prompt for password on CLI"""
+        prompt_mock = mocker.patch(
+            "rich.prompt.Prompt.ask", return_value=TEST_BACKUP_PASSWORD
+        )
+
+        runner = CliRunner()
+        path = os.path.join(get_artifact_folder(), "androidqf_encrypted")
+        result = runner.invoke(check_androidqf, [path])
+
+        assert prompt_mock.call_count == 1
+        assert result.exit_code == 0
+
+    def test_check_encrypted_backup_cli(self, mocker):
+        """Provide password as CLI argument"""
+        prompt_mock = mocker.patch(
+            "rich.prompt.Prompt.ask", return_value=TEST_BACKUP_PASSWORD
+        )
+
+        runner = CliRunner()
+        path = os.path.join(get_artifact_folder(), "androidqf_encrypted")
+        result = runner.invoke(
+            check_androidqf, ["--backup-password", TEST_BACKUP_PASSWORD, path]
+        )
+
+        assert prompt_mock.call_count == 0
+        assert result.exit_code == 0
+
+    def test_check_encrypted_backup_env(self, mocker):
+        """Provide password as environment variable"""
+        prompt_mock = mocker.patch(
+            "rich.prompt.Prompt.ask", return_value=TEST_BACKUP_PASSWORD
+        )
+
+        os.environ["MVT_ANDROID_BACKUP_PASSWORD"] = TEST_BACKUP_PASSWORD
+        runner = CliRunner()
+        path = os.path.join(get_artifact_folder(), "androidqf_encrypted")
+        result = runner.invoke(check_androidqf, [path])
+
+        assert prompt_mock.call_count == 0
+        assert result.exit_code == 0
+        del os.environ["MVT_ANDROID_BACKUP_PASSWORD"]

tests/test_check_android_backup.py

@@ -0,0 +1,73 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2023 Claudio Guarnieri.
+# 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 logging
+
+from click.testing import CliRunner
+
+from mvt.android.cli import check_backup
+
+from .utils import get_artifact_folder
+
+
+TEST_BACKUP_PASSWORD = "123456"
+
+
+class TestCheckAndroidBackupCommand:
+    def test_check_encrypted_backup_prompt_valid(self, mocker):
+        """Prompt for password on CLI"""
+        prompt_mock = mocker.patch(
+            "rich.prompt.Prompt.ask", return_value=TEST_BACKUP_PASSWORD
+        )
+        runner = CliRunner()
+        path = os.path.join(get_artifact_folder(), "androidqf_encrypted/backup.ab")
+        result = runner.invoke(check_backup, [path])
+
+        assert prompt_mock.call_count == 1
+        assert result.exit_code == 0
+
+    def test_check_encrypted_backup_cli(self, mocker):
+        """Provide password as CLI argument"""
+        prompt_mock = mocker.patch(
+            "rich.prompt.Prompt.ask", return_value=TEST_BACKUP_PASSWORD
+        )
+
+        runner = CliRunner()
+        path = os.path.join(get_artifact_folder(), "androidqf_encrypted/backup.ab")
+        result = runner.invoke(
+            check_backup, ["--backup-password", TEST_BACKUP_PASSWORD, path]
+        )
+
+        assert prompt_mock.call_count == 0
+        assert result.exit_code == 0
+
+    def test_check_encrypted_backup_cli_invalid(self, mocker, caplog):
+        """Provide password as CLI argument"""
+        runner = CliRunner()
+        path = os.path.join(get_artifact_folder(), "androidqf_encrypted/backup.ab")
+
+        with caplog.at_level(logging.CRITICAL):
+            result = runner.invoke(
+                check_backup, ["--backup-password", "invalid_password", path]
+            )
+
+        assert result.exit_code == 1
+        assert "Invalid backup password" in caplog.text
+
+    def test_check_encrypted_backup_env(self, mocker):
+        """Provide password as environment variable"""
+        prompt_mock = mocker.patch(
+            "rich.prompt.Prompt.ask", return_value=TEST_BACKUP_PASSWORD
+        )
+
+        os.environ["MVT_ANDROID_BACKUP_PASSWORD"] = TEST_BACKUP_PASSWORD
+        runner = CliRunner()
+        path = os.path.join(get_artifact_folder(), "androidqf_encrypted/backup.ab")
+        result = runner.invoke(check_backup, [path])
+
+        assert prompt_mock.call_count == 0
+        assert result.exit_code == 0
+        del os.environ["MVT_ANDROID_BACKUP_PASSWORD"]