Bumps version

* Remove duplicated detection logic from GetProp modules
* Deduplicate settings and processes
* Refactor detection in artifacts
* Improves Artifact class
---------

Co-authored-by: tek <tek@randhome.io>

.github/workflows/python-package.yml

@@ -28,7 +28,7 @@ jobs:
       run: |
         python -m pip install --upgrade setuptools
         python -m pip install --upgrade pip
-        python -m pip install flake8 pytest safety stix2 pytest-mock
+        python -m pip install flake8 pytest safety stix2 pytest-mock pytest-cov
         if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
         python -m pip install .
     - name: Lint with flake8
@@ -39,5 +39,10 @@ jobs:
         flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
     - name: Safety checks
       run: safety check
-    - name: Test with pytest
-      run: pytest
+    - name: Test with pytest and coverage
+      run: pytest --junitxml=pytest.xml --cov-report=term-missing:skip-covered --cov=mvt tests/ | tee pytest-coverage.txt
+    - name: Pytest coverage comment
+      uses: MishaKav/pytest-coverage-comment@main
+      with:
+        pytest-coverage-path: ./pytest-coverage.txt
+        junitxml-path: ./pytest.xml

.gitignore

@@ -50,6 +50,8 @@ coverage.xml
 *.py,cover
 .hypothesis/
 .pytest_cache/
+pytest-coverage.txt
+pytest.xml
 
 # Translations
 *.mo

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/artifacts/artifact.py

@@ -0,0 +1,9 @@
+# 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/
+from mvt.common.artifact import Artifact
+
+
+class AndroidArtifact(Artifact):
+    pass

mvt/android/artifacts/getprop.py

@@ -0,0 +1,59 @@
+# 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 re
+from typing import Dict, List
+
+from mvt.android.utils import warn_android_patch_level
+
+from .artifact import AndroidArtifact
+
+INTERESTING_PROPERTIES = [
+    "gsm.sim.operator.alpha",
+    "gsm.sim.operator.iso-country",
+    "persist.sys.timezone",
+    "ro.boot.serialno",
+    "ro.build.version.sdk",
+    "ro.build.version.security_patch",
+    "ro.product.cpu.abi",
+    "ro.product.locale",
+    "ro.product.vendor.manufacturer",
+    "ro.product.vendor.model",
+    "ro.product.vendor.name",
+]
+
+
+class GetProp(AndroidArtifact):
+    def parse(self, entry: str) -> None:
+        self.results: List[Dict[str, str]] = []
+        rxp = re.compile(r"\[(.+?)\]: \[(.+?)\]")
+
+        for line in entry.splitlines():
+            line = line.strip()
+            if line == "":
+                continue
+
+            matches = re.findall(rxp, line)
+            if not matches or len(matches[0]) != 2:
+                continue
+
+            entry = {"name": matches[0][0], "value": matches[0][1]}
+            self.results.append(entry)
+
+    def check_indicators(self) -> None:
+        for entry in self.results:
+            if entry["name"] in INTERESTING_PROPERTIES:
+                self.log.info("%s: %s", entry["name"], entry["value"])
+
+            if entry["name"] == "ro.build.version.security_patch":
+                warn_android_patch_level(entry["value"], self.log)
+
+        if not self.indicators:
+            return
+
+        for result in self.results:
+            ioc = self.indicators.check_android_property_name(result.get("name", ""))
+            if ioc:
+                result["matched_indicator"] = ioc
+                self.detected.append(result)

mvt/android/artifacts/processes.py

@@ -0,0 +1,69 @@
+# 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/
+from .artifact import AndroidArtifact
+
+
+class Processes(AndroidArtifact):
+    def parse(self, entry: str) -> None:
+        for line in entry.split("\n")[1:]:
+            proc = line.split()
+
+            # Skip empty lines
+            if len(proc) == 0:
+                continue
+
+            # Sometimes WCHAN is empty.
+            if len(proc) == 8:
+                proc = proc[:5] + [""] + proc[5:]
+
+            # Sometimes there is the security label.
+            if proc[0].startswith("u:r"):
+                label = proc[0]
+                proc = proc[1:]
+            else:
+                label = ""
+
+            # Sometimes there is no WCHAN.
+            if len(proc) < 9:
+                proc = proc[:5] + [""] + proc[5:]
+
+            self.results.append(
+                {
+                    "user": proc[0],
+                    "pid": int(proc[1]),
+                    "ppid": int(proc[2]),
+                    "virtual_memory_size": int(proc[3]),
+                    "resident_set_size": int(proc[4]),
+                    "wchan": proc[5],
+                    "aprocress": proc[6],
+                    "stat": proc[7],
+                    "proc_name": proc[8].strip("[]"),
+                    "label": label,
+                }
+            )
+
+    def check_indicators(self) -> None:
+        if not self.indicators:
+            return
+
+        for result in self.results:
+            proc_name = result.get("proc_name", "")
+            if not proc_name:
+                continue
+
+            # Skipping this process because of false positives.
+            if result["proc_name"] == "gatekeeperd":
+                continue
+
+            ioc = self.indicators.check_app_id(proc_name)
+            if ioc:
+                result["matched_indicator"] = ioc
+                self.detected.append(result)
+                continue
+
+            ioc = self.indicators.check_process(proc_name)
+            if ioc:
+                result["matched_indicator"] = ioc
+                self.detected.append(result)

mvt/android/artifacts/settings.py

@@ -0,0 +1,71 @@
+# 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/
+from .artifact import AndroidArtifact
+
+ANDROID_DANGEROUS_SETTINGS = [
+    {
+        "description": "disabled Google Play Services apps verification",
+        "key": "verifier_verify_adb_installs",
+        "safe_value": "1",
+    },
+    {
+        "description": "disabled Google Play Protect",
+        "key": "package_verifier_enable",
+        "safe_value": "1",
+    },
+    {
+        "description": "disabled Google Play Protect",
+        "key": "package_verifier_user_consent",
+        "safe_value": "1",
+    },
+    {
+        "description": "disabled Google Play Protect",
+        "key": "upload_apk_enable",
+        "safe_value": "1",
+    },
+    {
+        "description": "disabled confirmation of adb apps installation",
+        "key": "adb_install_need_confirm",
+        "safe_value": "1",
+    },
+    {
+        "description": "disabled sharing of security reports",
+        "key": "send_security_reports",
+        "safe_value": "1",
+    },
+    {
+        "description": "disabled sharing of crash logs with manufacturer",
+        "key": "samsung_errorlog_agree",
+        "safe_value": "1",
+    },
+    {
+        "description": "disabled applications errors reports",
+        "key": "send_action_app_error",
+        "safe_value": "1",
+    },
+    {
+        "description": "enabled installation of non Google Play apps",
+        "key": "install_non_market_apps",
+        "safe_value": "0",
+    },
+]
+
+
+class Settings(AndroidArtifact):
+    def check_indicators(self) -> None:
+        for namespace, settings in self.results.items():
+            for key, value in settings.items():
+                for danger in ANDROID_DANGEROUS_SETTINGS:
+                    # Check if one of the dangerous settings is using an unsafe
+                    # value (different than the one specified).
+                    if danger["key"] == key and danger["safe_value"] != value:
+                        self.log.warning(
+                            'Found suspicious "%s" setting "%s = %s" (%s)',
+                            namespace,
+                            key,
+                            value,
+                            danger["description"],
+                        )
+                        break

mvt/android/cli.py

@@ -9,11 +9,13 @@ import click
 
 from mvt.common.cmd_check_iocs import CmdCheckIOCS
 from mvt.common.help import (
+    HELP_MSG_ANDROID_BACKUP_PASSWORD,
     HELP_MSG_FAST,
     HELP_MSG_HASHES,
     HELP_MSG_IOC,
     HELP_MSG_LIST_MODULES,
     HELP_MSG_MODULE,
+    HELP_MSG_NONINTERACTIVE,
     HELP_MSG_OUTPUT,
     HELP_MSG_SERIAL,
     HELP_MSG_VERBOSE,
@@ -30,10 +32,12 @@ from .cmd_download_apks import DownloadAPKs
 from .modules.adb import ADB_MODULES
 from .modules.adb.packages import Packages
 from .modules.backup import BACKUP_MODULES
+from .modules.backup.helpers import cli_load_android_backup_password
 from .modules.bugreport import BUGREPORT_MODULES
 
 init_logging()
 log = logging.getLogger("mvt")
+
 CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])
 
 
@@ -125,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)
@@ -141,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,
@@ -234,14 +255,33 @@ def check_bugreport(ctx, iocs, output, list_modules, module, verbose, bugreport_
 )
 @click.option("--output", "-o", type=click.Path(exists=False), help=HELP_MSG_OUTPUT)
 @click.option("--list-modules", "-l", is_flag=True, help=HELP_MSG_LIST_MODULES)
+@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.argument("BACKUP_PATH", type=click.Path(exists=True))
 @click.pass_context
-def check_backup(ctx, iocs, output, list_modules, verbose, backup_path):
+def check_backup(
+    ctx,
+    iocs,
+    output,
+    list_modules,
+    non_interactive,
+    backup_password,
+    verbose,
+    backup_path,
+):
     set_verbose_logging(verbose)
+
     # Always generate hashes as backups are generally small.
     cmd = CmdAndroidCheckBackup(
-        target_path=backup_path, results_path=output, ioc_files=iocs, hashes=True
+        target_path=backup_path,
+        results_path=output,
+        ioc_files=iocs,
+        hashes=True,
+        module_options={
+            "interactive": not non_interactive,
+            "backup_password": cli_load_android_backup_password(log, backup_password),
+        },
     )
 
     if list_modules:
@@ -279,19 +319,35 @@ def check_backup(ctx, iocs, output, list_modules, verbose, backup_path):
 @click.option("--list-modules", "-l", is_flag=True, help=HELP_MSG_LIST_MODULES)
 @click.option("--module", "-m", help=HELP_MSG_MODULE)
 @click.option("--hashes", "-H", is_flag=True, help=HELP_MSG_HASHES)
+@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.argument("ANDROIDQF_PATH", type=click.Path(exists=True))
 @click.pass_context
 def check_androidqf(
-    ctx, iocs, output, list_modules, module, hashes, verbose, androidqf_path
+    ctx,
+    iocs,
+    output,
+    list_modules,
+    module,
+    hashes,
+    non_interactive,
+    backup_password,
+    verbose,
+    androidqf_path,
 ):
     set_verbose_logging(verbose)
+
     cmd = CmdAndroidCheckAndroidQF(
         target_path=androidqf_path,
         results_path=output,
         ioc_files=iocs,
         module_name=module,
         hashes=hashes,
+        module_options={
+            "interactive": not non_interactive,
+            "backup_password": cli_load_android_backup_password(log, backup_password),
+        },
     )
 
     if list_modules:

mvt/android/cmd_check_androidqf.py

@@ -4,7 +4,10 @@
 #   https://license.mvt.re/1.1/
 
 import logging
-from typing import Optional
+import os
+import zipfile
+from pathlib import Path
+from typing import List, Optional
 
 from mvt.common.command import Command
 
@@ -37,3 +40,28 @@ class CmdAndroidCheckAndroidQF(Command):
 
         self.name = "check-androidqf"
         self.modules = ANDROIDQF_MODULES
+
+        self.format: Optional[str] = None
+        self.archive: Optional[zipfile.ZipFile] = None
+        self.files: List[str] = []
+
+    def init(self):
+        if os.path.isdir(self.target_path):
+            self.format = "dir"
+            parent_path = Path(self.target_path).absolute().parent.as_posix()
+            target_abs_path = os.path.abspath(self.target_path)
+            for root, subdirs, subfiles in os.walk(target_abs_path):
+                for fname in subfiles:
+                    file_path = os.path.relpath(os.path.join(root, fname), parent_path)
+                    self.files.append(file_path)
+        elif os.path.isfile(self.target_path):
+            self.format = "zip"
+            self.archive = zipfile.ZipFile(self.target_path)
+            self.files = self.archive.namelist()
+
+    def module_init(self, module):
+        if self.format == "zip":
+            module.from_zip_file(self.archive, self.files)
+        else:
+            parent_path = Path(self.target_path).absolute().parent.as_posix()
+            module.from_folder(parent_path, self.files)

mvt/android/cmd_check_backup.py

@@ -11,9 +11,8 @@ 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.modules.backup.helpers import prompt_or_load_android_backup_password
 from mvt.android.parsers.backup import (
     AndroidBackupParsingError,
     InvalidBackupPassword,
@@ -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,9 +22,9 @@ from adb_shell.exceptions import (
     UsbDeviceNotFoundError,
     UsbReadFailedError,
 )
-from rich.prompt import Prompt
 from usb1 import USBErrorAccess, USBErrorBusy
 
+from mvt.android.modules.backup.helpers import prompt_or_load_android_backup_password
 from mvt.android.parsers.backup import (
     InvalidBackupPassword,
     parse_ab_header,
@@ -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/adb/getprop.py

@@ -4,15 +4,14 @@
 #   https://license.mvt.re/1.1/
 
 import logging
-from datetime import datetime, timedelta
 from typing import Optional
 
-from mvt.android.parsers import parse_getprop
+from mvt.android.artifacts.getprop import GetProp as GetPropArtifact
 
 from .base import AndroidExtraction
 
 
-class Getprop(AndroidExtraction):
+class Getprop(GetPropArtifact, AndroidExtraction):
     """This module extracts device properties from getprop command."""
 
     def __init__(
@@ -35,33 +34,10 @@ class Getprop(AndroidExtraction):
 
         self.results = {} if not results else results
 
-    def check_indicators(self) -> None:
-        if not self.indicators:
-            return
-
-        for result in self.results:
-            ioc = self.indicators.check_android_property_name(result.get("name", ""))
-            if ioc:
-                result["matched_indicator"] = ioc
-                self.detected.append(result)
-
     def run(self) -> None:
         self._adb_connect()
         output = self._adb_command("getprop")
         self._adb_disconnect()
 
-        self.results = parse_getprop(output)
-
-        # Alert if phone is outdated.
-        for entry in self.results:
-            if entry.get("name", "") != "ro.build.version.security_patch":
-                continue
-            patch_date = datetime.strptime(entry["value"], "%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)",
-                    entry["value"],
-                )
-
+        self.parse(output)
         self.log.info("Extracted %d Android system properties", len(self.results))

mvt/android/modules/adb/processes.py

@@ -6,10 +6,12 @@
 import logging
 from typing import Optional
 
+from mvt.android.artifacts.processes import Processes as ProcessesArtifact
+
 from .base import AndroidExtraction
 
 
-class Processes(AndroidExtraction):
+class Processes(ProcessesArtifact, AndroidExtraction):
     """This module extracts details on running processes."""
 
     def __init__(
@@ -30,61 +32,11 @@ class Processes(AndroidExtraction):
             results=results,
         )
 
-    def check_indicators(self) -> None:
-        if not self.indicators:
-            return
-
-        for result in self.results:
-            proc_name = result.get("proc_name", "")
-            if not proc_name:
-                continue
-
-            # Skipping this process because of false positives.
-            if result["proc_name"] == "gatekeeperd":
-                continue
-
-            ioc = self.indicators.check_app_id(proc_name)
-            if ioc:
-                result["matched_indicator"] = ioc
-                self.detected.append(result)
-                continue
-
-            ioc = self.indicators.check_process(proc_name)
-            if ioc:
-                result["matched_indicator"] = ioc
-                self.detected.append(result)
-
     def run(self) -> None:
         self._adb_connect()
 
         output = self._adb_command("ps -A")
-
-        for line in output.splitlines()[1:]:
-            line = line.strip()
-            if line == "":
-                continue
-
-            fields = line.split()
-            proc = {
-                "user": fields[0],
-                "pid": fields[1],
-                "parent_pid": fields[2],
-                "vsize": fields[3],
-                "rss": fields[4],
-            }
-
-            # Sometimes WCHAN is empty, so we need to re-align output fields.
-            if len(fields) == 8:
-                proc["wchan"] = ""
-                proc["pc"] = fields[5]
-                proc["name"] = fields[7]
-            elif len(fields) == 9:
-                proc["wchan"] = fields[5]
-                proc["pc"] = fields[6]
-                proc["name"] = fields[8]
-
-            self.results.append(proc)
-
+        self.parse(output)
         self._adb_disconnect()
 
         self.log.info("Extracted records on a total of %d processes", len(self.results))

mvt/android/modules/adb/settings.py

@@ -6,58 +6,12 @@
 import logging
 from typing import Optional
 
+from mvt.android.artifacts.settings import Settings as SettingsArtifact
+
 from .base import AndroidExtraction
 
-ANDROID_DANGEROUS_SETTINGS = [
-    {
-        "description": "disabled Google Play Services apps verification",
-        "key": "verifier_verify_adb_installs",
-        "safe_value": "1",
-    },
-    {
-        "description": "disabled Google Play Protect",
-        "key": "package_verifier_enable",
-        "safe_value": "1",
-    },
-    {
-        "description": "disabled Google Play Protect",
-        "key": "package_verifier_user_consent",
-        "safe_value": "1",
-    },
-    {
-        "description": "disabled Google Play Protect",
-        "key": "upload_apk_enable",
-        "safe_value": "1",
-    },
-    {
-        "description": "disabled confirmation of adb apps installation",
-        "key": "adb_install_need_confirm",
-        "safe_value": "1",
-    },
-    {
-        "description": "disabled sharing of security reports",
-        "key": "send_security_reports",
-        "safe_value": "1",
-    },
-    {
-        "description": "disabled sharing of crash logs with manufacturer",
-        "key": "samsung_errorlog_agree",
-        "safe_value": "1",
-    },
-    {
-        "description": "disabled applications errors reports",
-        "key": "send_action_app_error",
-        "safe_value": "1",
-    },
-    {
-        "description": "enabled installation of non Google Play apps",
-        "key": "install_non_market_apps",
-        "safe_value": "0",
-    },
-]
 
-
-class Settings(AndroidExtraction):
+class Settings(SettingsArtifact, AndroidExtraction):
     """This module extracts Android system settings."""
 
     def __init__(
@@ -80,21 +34,6 @@ class Settings(AndroidExtraction):
 
         self.results = {} if not results else results
 
-    def check_indicators(self) -> None:
-        for _, settings in self.results.items():
-            for key, value in settings.items():
-                for danger in ANDROID_DANGEROUS_SETTINGS:
-                    # Check if one of the dangerous settings is using an unsafe
-                    # value (different than the one specified).
-                    if danger["key"] == key and danger["safe_value"] != value:
-                        self.log.warning(
-                            'Found suspicious setting "%s = %s" (%s)',
-                            key,
-                            value,
-                            danger["description"],
-                        )
-                        break
-
     def run(self) -> None:
         self._adb_connect()
 

mvt/android/modules/androidqf/base.py

@@ -6,6 +6,7 @@
 import fnmatch
 import logging
 import os
+import zipfile
 from typing import Any, Dict, List, Optional, Union
 
 from mvt.common.module import MVTModule
@@ -31,13 +32,28 @@ class AndroidQFModule(MVTModule):
             log=log,
             results=results,
         )
+        self._path: str = target_path
+        self.files: List[str] = []
+        self.archive: Optional[zipfile.ZipFile] = None
 
-        self._path = target_path
-        self._files = []
+    def from_folder(self, parent_path: str, files: List[str]):
+        self.parent_path = parent_path
+        self.files = files
 
-        for root, dirs, files in os.walk(target_path):
-            for name in files:
-                self._files.append(os.path.join(root, name))
+    def from_zip_file(self, archive: zipfile.ZipFile, files: List[str]):
+        self.archive = archive
+        self.files = files
 
-    def _get_files_by_pattern(self, pattern):
-        return fnmatch.filter(self._files, pattern)
+    def _get_files_by_pattern(self, pattern: str):
+        return fnmatch.filter(self.files, pattern)
+
+    def _get_file_content(self, file_path):
+        if self.archive:
+            handle = self.archive.open(file_path)
+        else:
+            handle = open(os.path.join(self.parent_path, file_path), "rb")
+
+        data = handle.read()
+        handle.close()
+
+        return data

mvt/android/modules/androidqf/dumpsys_accessibility.py

@@ -49,21 +49,21 @@ class DumpsysAccessibility(AndroidQFModule):
 
         lines = []
         in_accessibility = False
-        with open(dumpsys_file[0]) as handle:
-            for line in handle:
-                if line.strip().startswith("DUMP OF SERVICE accessibility:"):
-                    in_accessibility = True
-                    continue
+        data = self._get_file_content(dumpsys_file[0])
+        for line in data.decode("utf-8").split("\n"):
+            if line.strip().startswith("DUMP OF SERVICE accessibility:"):
+                in_accessibility = True
+                continue
 
-                if not in_accessibility:
-                    continue
+            if not in_accessibility:
+                continue
 
-                if line.strip().startswith(
-                    "-------------------------------------------------------------------------------"
-                ):  # pylint: disable=line-too-long
-                    break
+            if line.strip().startswith(
+                "-------------------------------------------------------------------------------"
+            ):  # pylint: disable=line-too-long
+                break
 
-                lines.append(line.rstrip())
+            lines.append(line.rstrip())
 
         self.results = parse_dumpsys_accessibility("\n".join(lines))
 

mvt/android/modules/androidqf/dumpsys_activities.py

@@ -52,21 +52,21 @@ class DumpsysActivities(AndroidQFModule):
 
         lines = []
         in_package = False
-        with open(dumpsys_file[0]) as handle:
-            for line in handle:
-                if line.strip() == "DUMP OF SERVICE package:":
-                    in_package = True
-                    continue
+        data = self._get_file_content(dumpsys_file[0])
+        for line in data.decode("utf-8").split("\n"):
+            if line.strip() == "DUMP OF SERVICE package:":
+                in_package = True
+                continue
 
-                if not in_package:
-                    continue
+            if not in_package:
+                continue
 
-                if line.strip().startswith(
-                    "------------------------------------------------------------------------------"
-                ):  # pylint: disable=line-too-long
-                    break
+            if line.strip().startswith(
+                "------------------------------------------------------------------------------"
+            ):  # pylint: disable=line-too-long
+                break
 
-                lines.append(line.rstrip())
+            lines.append(line.rstrip())
 
         self.results = parse_dumpsys_activity_resolver_table("\n".join(lines))
 

mvt/android/modules/androidqf/dumpsys_appops.py

@@ -76,19 +76,19 @@ class DumpsysAppops(AndroidQFModule):
 
         lines = []
         in_package = False
-        with open(dumpsys_file[0]) as handle:
-            for line in handle:
-                if line.startswith("DUMP OF SERVICE appops:"):
-                    in_package = True
-                    continue
+        data = self._get_file_content(dumpsys_file[0])
+        for line in data.decode("utf-8").split("\n"):
+            if line.startswith("DUMP OF SERVICE appops:"):
+                in_package = True
+                continue
 
-                if in_package:
-                    if line.startswith(
-                        "-------------------------------------------------------------------------------"
-                    ):  # pylint: disable=line-too-long
-                        break
+            if in_package:
+                if line.startswith(
+                    "-------------------------------------------------------------------------------"
+                ):  # pylint: disable=line-too-long
+                    break
 
-                    lines.append(line.rstrip())
+                lines.append(line.rstrip())
 
         self.results = parse_dumpsys_appops("\n".join(lines))
         self.log.info("Identified %d applications in AppOps Manager", len(self.results))

mvt/android/modules/androidqf/dumpsys_packages.py

@@ -78,13 +78,12 @@ class DumpsysPackages(AndroidQFModule):
             self.log.info("Dumpsys file not found")
             return
 
-        with open(dumpsys_file[0]) as handle:
-            data = handle.read().split("\n")
+        data = self._get_file_content(dumpsys_file[0])
 
         package = []
         in_service = False
         in_package_list = False
-        for line in data:
+        for line in data.decode("utf-8").split("\n"):
             if line.strip().startswith("DUMP OF SERVICE package:"):
                 in_service = True
                 continue

mvt/android/modules/androidqf/dumpsys_receivers.py

@@ -86,21 +86,21 @@ class DumpsysReceivers(AndroidQFModule):
 
         in_receivers = False
         lines = []
-        with open(dumpsys_file[0]) as handle:
-            for line in handle:
-                if line.strip() == "DUMP OF SERVICE package:":
-                    in_receivers = True
-                    continue
+        data = self._get_file_content(dumpsys_file[0])
+        for line in data.decode("utf-8").split("\n"):
+            if line.strip() == "DUMP OF SERVICE package:":
+                in_receivers = True
+                continue
 
-                if not in_receivers:
-                    continue
+            if not in_receivers:
+                continue
 
-                if line.strip().startswith(
-                    "------------------------------------------------------------------------------"
-                ):  # pylint: disable=line-too-long
-                    break
+            if line.strip().startswith(
+                "------------------------------------------------------------------------------"
+            ):  # pylint: disable=line-too-long
+                break
 
-                lines.append(line.rstrip())
+            lines.append(line.rstrip())
 
         self.results = parse_dumpsys_receiver_resolver_table("\n".join(lines))
 

mvt/android/modules/androidqf/getprop.py

@@ -4,29 +4,14 @@
 #   https://license.mvt.re/1.1/
 
 import logging
-from datetime import datetime, timedelta
 from typing import Optional
 
-from mvt.android.parsers.getprop import parse_getprop
+from mvt.android.artifacts.getprop import GetProp as GetPropArtifact
 
 from .base import AndroidQFModule
 
-INTERESTING_PROPERTIES = [
-    "gsm.sim.operator.alpha",
-    "gsm.sim.operator.iso-country",
-    "persist.sys.timezone",
-    "ro.boot.serialno",
-    "ro.build.version.sdk",
-    "ro.build.version.security_patch",
-    "ro.product.cpu.abi",
-    "ro.product.locale",
-    "ro.product.vendor.manufacturer",
-    "ro.product.vendor.model",
-    "ro.product.vendor.name",
-]
 
-
-class Getprop(AndroidQFModule):
+class Getprop(GetPropArtifact, AndroidQFModule):
     """This module extracts data from get properties."""
 
     def __init__(
@@ -48,37 +33,13 @@ class Getprop(AndroidQFModule):
         )
         self.results = []
 
-    def check_indicators(self) -> None:
-        if not self.indicators:
-            return
-
-        for result in self.results:
-            ioc = self.indicators.check_android_property_name(result.get("name", ""))
-            if ioc:
-                result["matched_indicator"] = ioc
-                self.detected.append(result)
-
     def run(self) -> None:
         getprop_files = self._get_files_by_pattern("*/getprop.txt")
         if not getprop_files:
             self.log.info("getprop.txt file not found")
             return
 
-        with open(getprop_files[0]) as f:
-            data = f.read()
-
-        self.results = parse_getprop(data)
-        for entry in self.results:
-            if entry["name"] in INTERESTING_PROPERTIES:
-                self.log.info("%s: %s", entry["name"], entry["value"])
-            if entry["name"] == "ro.build.version.security_patch":
-                last_patch = datetime.strptime(entry["value"], "%Y-%m-%d")
-                if (datetime.now() - last_patch) > timedelta(days=6 * 31):
-                    self.log.warning(
-                        "This phone has not received security "
-                        "updates for more than six months "
-                        "(last update: %s)",
-                        entry["value"],
-                    )
+        data = self._get_file_content(getprop_files[0]).decode("utf-8")
 
+        self.parse(data)
         self.log.info("Extracted a total of %d properties", len(self.results))

mvt/android/modules/androidqf/processes.py

@@ -6,10 +6,12 @@
 import logging
 from typing import Optional
 
+from mvt.android.artifacts.processes import Processes as ProcessesArtifact
+
 from .base import AndroidQFModule
 
 
-class Processes(AndroidQFModule):
+class Processes(ProcessesArtifact, AndroidQFModule):
     """This module analyse running processes"""
 
     def __init__(
@@ -30,70 +32,10 @@ class Processes(AndroidQFModule):
             results=results,
         )
 
-    def check_indicators(self) -> None:
-        if not self.indicators:
-            return
-
-        for result in self.results:
-            proc_name = result.get("proc_name", "")
-            if not proc_name:
-                continue
-
-            # Skipping this process because of false positives.
-            if result["proc_name"] == "gatekeeperd":
-                continue
-
-            ioc = self.indicators.check_app_id(proc_name)
-            if ioc:
-                result["matched_indicator"] = ioc
-                self.detected.append(result)
-                continue
-
-            ioc = self.indicators.check_process(proc_name)
-            if ioc:
-                result["matched_indicator"] = ioc
-                self.detected.append(result)
-
-    def _parse_ps(self, data):
-        for line in data.split("\n")[1:]:
-            proc = line.split()
-
-            # Sometimes WCHAN is empty.
-            if len(proc) == 8:
-                proc = proc[:5] + [""] + proc[5:]
-
-            # Sometimes there is the security label.
-            if proc[0].startswith("u:r"):
-                label = proc[0]
-                proc = proc[1:]
-            else:
-                label = ""
-
-            # Sometimes there is no WCHAN.
-            if len(proc) < 9:
-                proc = proc[:5] + [""] + proc[5:]
-
-            self.results.append(
-                {
-                    "user": proc[0],
-                    "pid": int(proc[1]),
-                    "ppid": int(proc[2]),
-                    "virtual_memory_size": int(proc[3]),
-                    "resident_set_size": int(proc[4]),
-                    "wchan": proc[5],
-                    "aprocress": proc[6],
-                    "stat": proc[7],
-                    "proc_name": proc[8].strip("[]"),
-                    "label": label,
-                }
-            )
-
     def run(self) -> None:
         ps_files = self._get_files_by_pattern("*/ps.txt")
         if not ps_files:
             return
 
-        with open(ps_files[0]) as handle:
-            self._parse_ps(handle.read())
-
+        self.parse(self._get_file_content(ps_files[0]).decode("utf-8"))
         self.log.info("Identified %d running processes", len(self.results))

mvt/android/modules/androidqf/settings.py

@@ -6,12 +6,12 @@
 import logging
 from typing import Optional
 
-from mvt.android.modules.adb.settings import ANDROID_DANGEROUS_SETTINGS
+from mvt.android.artifacts.settings import Settings as SettingsArtifact
 
 from .base import AndroidQFModule
 
 
-class Settings(AndroidQFModule):
+class Settings(SettingsArtifact, AndroidQFModule):
     """This module analyse setting files"""
 
     def __init__(
@@ -38,29 +38,18 @@ class Settings(AndroidQFModule):
             namespace = setting_file[setting_file.rfind("_") + 1 : -4]
 
             self.results[namespace] = {}
+            data = self._get_file_content(setting_file)
+            for line in data.decode("utf-8").split("\n"):
+                line = line.strip()
+                try:
+                    key, value = line.split("=", 1)
+                except ValueError:
+                    continue
 
-            with open(setting_file) as handle:
-                for line in handle:
-                    line = line.strip()
-                    try:
-                        key, value = line.split("=", 1)
-                    except ValueError:
-                        continue
-
-                    try:
-                        self.results[namespace][key] = value
-                    except IndexError:
-                        continue
-
-                    for danger in ANDROID_DANGEROUS_SETTINGS:
-                        if danger["key"] == key and danger["safe_value"] != value:
-                            self.log.warning(
-                                'Found suspicious setting "%s = %s" (%s)',
-                                key,
-                                value,
-                                danger["description"],
-                            )
-                            break
+                try:
+                    self.results[namespace][key] = value
+                except IndexError:
+                    continue
 
         self.log.info(
             "Identified %d settings", sum([len(val) for val in self.results.values()])

mvt/android/modules/androidqf/sms.py

@@ -3,10 +3,10 @@
 # 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 logging
 from typing import Optional
 
+from mvt.android.modules.backup.helpers import prompt_or_load_android_backup_password
 from mvt.android.parsers.backup import (
     AndroidBackupParsingError,
     InvalidBackupPassword,
@@ -58,7 +58,13 @@ class SMS(AndroidQFModule):
 
         password = None
         if header["encryption"] != "none":
-            password = getpass.getpass(prompt="Backup Password: ", stream=None)
+            password = prompt_or_load_android_backup_password(
+                self.log, self.module_options
+            )
+            if not password:
+                self.log.critical("No backup password provided.")
+                return
+
         try:
             tardata = parse_backup_file(data, password=password)
         except InvalidBackupPassword:
@@ -90,8 +96,5 @@ class SMS(AndroidQFModule):
             self.log.info("No backup data found")
             return
 
-        with open(files[0], "rb") as handle:
-            data = handle.read()
-
-        self.parse_backup(data)
+        self.parse_backup(self._get_file_content(files[0]))
         self.log.info("Identified %d SMS in backup data", len(self.results))

mvt/android/modules/backup/helpers.py

@@ -0,0 +1,60 @@
+# 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

mvt/android/modules/bugreport/getprop.py

@@ -4,15 +4,14 @@
 #   https://license.mvt.re/1.1/
 
 import logging
-from datetime import datetime, timedelta
 from typing import Optional
 
-from mvt.android.parsers import parse_getprop
+from mvt.android.artifacts.getprop import GetProp as GetPropArtifact
 
 from .base import BugReportModule
 
 
-class Getprop(BugReportModule):
+class Getprop(GetPropArtifact, BugReportModule):
     """This module extracts device properties from getprop command."""
 
     def __init__(
@@ -33,7 +32,7 @@ class Getprop(BugReportModule):
             results=results,
         )
 
-        self.results = {} if not results else results
+        self.results = [] if not results else results
 
     def run(self) -> None:
         content = self._get_dumpstate_file()
@@ -60,18 +59,5 @@ class Getprop(BugReportModule):
 
             lines.append(line)
 
-        self.results = parse_getprop("\n".join(lines))
-
-        # Alert if phone is outdated.
-        for entry in self.results:
-            if entry["name"] == "ro.build.version.security_patch":
-                security_patch = entry["value"]
-                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.parse("\n".join(lines))
         self.log.info("Extracted %d Android system properties", len(self.results))

mvt/android/parsers/__init__.py

@@ -12,4 +12,3 @@ from .dumpsys import (
     parse_dumpsys_dbinfo,
     parse_dumpsys_receiver_resolver_table,
 )
-from .getprop import parse_getprop

mvt/android/parsers/getprop.py

@@ -1,26 +0,0 @@
-# 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 re
-from typing import Dict, List
-
-
-def parse_getprop(output: str) -> List[Dict[str, str]]:
-    results = []
-    rxp = re.compile(r"\[(.+?)\]: \[(.+?)\]")
-
-    for line in output.splitlines():
-        line = line.strip()
-        if line == "":
-            continue
-
-        matches = re.findall(rxp, line)
-        if not matches or len(matches[0]) != 2:
-            continue
-
-        entry = {"name": matches[0][0], "value": matches[0][1]}
-        results.append(entry)
-
-    return results

mvt/android/utils.py

@@ -0,0 +1,19 @@
+# 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/
+from datetime import datetime, timedelta
+
+
+def warn_android_patch_level(patch_level: str, log) -> bool:
+    """Alert if Android patch level out-of-date"""
+    patch_date = datetime.strptime(patch_level, "%Y-%m-%d")
+    if (datetime.now() - patch_date) > timedelta(days=6 * 31):
+        log.warning(
+            "This phone has not received security updates "
+            "for more than six months (last update: %s)",
+            patch_level,
+        )
+        return True
+
+    return False

mvt/common/artifact.py

@@ -0,0 +1,28 @@
+# 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/
+
+
+class Artifact:
+    """
+    Main artifact class
+    """
+
+    def __init__(self, *args, **kwargs):
+        self.results = []
+        self.detected = []
+        self.indicators = None
+        super().__init__(*args, **kwargs)
+
+    def parse(self, entry: str):
+        """
+        Parse the artifact, adds the parsed information to self.results
+        """
+        raise NotImplementedError
+
+    def check_indicators(self) -> None:
+        """Check the results of this module against a provided list of
+        indicators coming from self.indicators
+        """
+        raise NotImplementedError

mvt/common/help.py

@@ -9,6 +9,8 @@ HELP_MSG_IOC = "Path to indicators file (can be invoked multiple time)"
 HELP_MSG_FAST = "Avoid running time/resource consuming features"
 HELP_MSG_LIST_MODULES = "Print list of available modules and exit"
 HELP_MSG_MODULE = "Name of a single module you would like to run instead of all"
+HELP_MSG_NONINTERACTIVE = "Don't ask interactive questions during processing"
+HELP_MSG_ANDROID_BACKUP_PASSWORD = "The backup password to use for an Android backup"
 HELP_MSG_HASHES = "Generate hashes of all the files analyzed"
 HELP_MSG_VERBOSE = "Verbose mode"