Bumps version

Update applications.py to add extra valid source

.github/workflows/black.yml

@@ -0,0 +1,11 @@
+name: Black
+on: [push]
+
+jobs:
+  black:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - uses: psf/black@stable
+        with:
+          options: "--check"

.github/workflows/python-package.yml

@@ -16,19 +16,19 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ['3.8', '3.9', '3.10']
+        python-version: ['3.8', '3.9', '3.10']  # , '3.11']
 
     steps:
-    - uses: actions/checkout@v2
+    - uses: actions/checkout@v3
     - name: Set up Python ${{ matrix.python-version }}
-      uses: actions/setup-python@v2
+      uses: actions/setup-python@v4
       with:
         python-version: ${{ matrix.python-version }}
     - name: Install dependencies
       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,11 @@ 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
+      if: github.event_name == 'pull_request'
+      with:
+        pytest-coverage-path: ./pytest-coverage.txt
+        junitxml-path: ./pytest.xml

.github/workflows/ruff.yml

@@ -1,21 +1,19 @@
 name: Ruff
-on: [push]
-
+on:
+  push:
+    branches: [ main ]
+  pull_request:
+    branches: [ main ]
 jobs:
   ruff_py3:
     name: Ruff syntax check
     runs-on: ubuntu-latest
     steps:
-      - name: Setup Python
-        uses: actions/setup-python@v1
-        with:
-          python-version: 3.9
-          architecture: x64
       - name: Checkout
         uses: actions/checkout@master
       - name: Install Dependencies
         run: |
-          pip install ruff
+          pip install --user ruff
       - name: ruff
         run: |
-          ruff check .
+          ruff --format=github .

.github/workflows/scripts/update-ios-releases.py

@@ -2,8 +2,8 @@
 Python script to download the Apple RSS feed and parse it.
 """
 
-import os
 import json
+import os
 import urllib.request
 from xml.dom.minidom import parseString
 
@@ -12,7 +12,7 @@ from packaging import version
 
 def download_apple_rss(feed_url):
     with urllib.request.urlopen(feed_url) as f:
-        rss_feed = f.read().decode('utf-8')
+        rss_feed = f.read().decode("utf-8")
     print("Downloaded RSS feed from Apple.")
     return rss_feed
 
@@ -27,7 +27,10 @@ def parse_latest_ios_versions(rss_feed_text):
             continue
 
         import re
-        build_match = re.match(r"iOS (?P<version>[\d\.]+) (?P<beta>beta )?(\S*)?\((?P<build>.*)\)", title)
+
+        build_match = re.match(
+            r"iOS (?P<version>[\d\.]+) (?P<beta>beta )?(\S*)?\((?P<build>.*)\)", title
+        )
         if not build_match:
             print("Could not parse iOS build:", title)
             continue
@@ -62,16 +65,22 @@ def update_mvt(mvt_checkout_path, latest_ios_versions):
         print("No new iOS versions found.")
     else:
         print("Found {} new iOS versions.".format(new_entry_count))
-        new_version_list = sorted(current_versions, key=lambda x: version.Version(x["version"]))
+        new_version_list = sorted(
+            current_versions, key=lambda x: version.Version(x["version"])
+        )
         with open(version_path, "w") as version_file:
             json.dump(new_version_list, version_file, indent=4)
 
 
 def main():
     print("Downloading RSS feed...")
-    mvt_checkout_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../"))
+    mvt_checkout_path = os.path.abspath(
+        os.path.join(os.path.dirname(__file__), "../../../")
+    )
 
-    rss_feed = download_apple_rss("https://developer.apple.com/news/releases/rss/releases.rss")
+    rss_feed = download_apple_rss(
+        "https://developer.apple.com/news/releases/rss/releases.rss"
+    )
     latest_ios_version = parse_latest_ios_versions(rss_feed)
     update_mvt(mvt_checkout_path, latest_ios_version)
 

.gitignore

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

Makefile

@@ -2,8 +2,10 @@ PWD = $(shell pwd)
 
 check:
 	flake8
-	pytest -q
 	ruff check -q .
+	black --check .
+	pytest -q
+
 
 clean:
 	rm -rf $(PWD)/build $(PWD)/dist $(PWD)/mvt.egg-info

README.md

@@ -11,10 +11,24 @@
 
 Mobile Verification Toolkit (MVT) is a collection of utilities to simplify and automate the process of gathering forensic traces helpful to identify a potential compromise of Android and iOS devices.
 
-It has been developed and released by the [Amnesty International Security Lab](https://www.amnesty.org/en/tech/) in July 2021 in the context of the [Pegasus project](https://forbiddenstories.org/about-the-pegasus-project/) along with [a technical forensic methodology and forensic evidence](https://www.amnesty.org/en/latest/research/2021/07/forensic-methodology-report-how-to-catch-nso-groups-pegasus/).
+It has been developed and released by the [Amnesty International Security Lab](https://www.amnesty.org/en/tech/) in July 2021 in the context of the [Pegasus Project](https://forbiddenstories.org/about-the-pegasus-project/) along with [a technical forensic methodology](https://www.amnesty.org/en/latest/research/2021/07/forensic-methodology-report-how-to-catch-nso-groups-pegasus/). It continues to be maintained by Amnesty International and other contributors.
 
-*Warning*: MVT is a forensic research tool intended for technologists and investigators. Using it requires understanding the basics of forensic analysis and using command-line tools. This is not intended for end-user self-assessment. If you are concerned with the security of your device please seek expert assistance.
+> **Note**
+> MVT is a forensic research tool intended for technologists and investigators. It requires understanding digital forensics and using command-line tools. This is not intended for end-user self-assessment. If you are concerned with the security of your device please seek reputable expert assistance.
+>
 
+### Indicators of Compromise
+
+MVT supports using public [indicators of compromise (IOCs)](https://github.com/mvt-project/mvt-indicators) to scan mobile devices for potential traces of targeting or infection by known spyware campaigns. This includes IOCs published by [Amnesty International](https://github.com/AmnestyTech/investigations/) and other  research groups.
+
+> **Warning**
+> Public indicators of compromise are insufficient to determine that a device is "clean", and not targeted with a particular spyware tool. Reliance on public indicators alone can miss recent forensic traces and give a false sense of security.
+>
+> Reliable and comprehensive digital forensic support and triage requires access to non-public indicators, research and threat intelligence.
+>
+>Such support is available to civil society through [Amnesty International's Security Lab](https://www.amnesty.org/en/tech/) or through our forensic partnership with [Access Now’s Digital Security Helpline](https://www.accessnow.org/help/).
+
+More information about using indicators of compromise with MVT is available in the [documentation](https://docs.mvt.re/en/latest/iocs/).
 
 ## Installation
 

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.
 

docs/development.md

@@ -0,0 +1,27 @@
+# Development
+
+The Mobile Verification Toolkit team welcomes contributions of new forensic modules or other contributions which help improve the software.
+
+## Testing
+
+MVT uses `pytest` for unit and integration tests. Code style consistency is maintained with `flake8`, `ruff` and `black`. All can
+be run automatically with:
+
+```bash
+make check
+```
+
+Run these tests before making new commits or opening pull requests.
+
+## Profiling
+
+Some MVT modules extract and process significant amounts of data during the analysis process or while checking results against known indicators. Care must be
+take to avoid inefficient code paths as we add new modules.
+
+MVT modules can be profiled with Python built-in `cProfile` by setting the `MVT_PROFILE` environment variable.
+
+```bash
+MVT_PROFILE=1 dev/mvt-ios check-backup test_backup
+```
+
+Open an issue or PR if you are encountering significant performance issues when analyzing a device with MVT.

docs/index.md

@@ -6,6 +6,9 @@
 
 Mobile Verification Toolkit (MVT) is a tool to facilitate the [consensual forensic analysis](introduction.md#consensual-forensics) of Android and iOS devices, for the purpose of identifying traces of compromise.
 
+It has been developed and released by the [Amnesty International Security Lab](https://www.amnesty.org/en/tech/) in July 2021 in the context of the [Pegasus Project](https://forbiddenstories.org/about-the-pegasus-project/) along with [a technical forensic methodology](https://www.amnesty.org/en/latest/research/2021/07/forensic-methodology-report-how-to-catch-nso-groups-pegasus/). It continues to be maintained by Amnesty International and other contributors.
+
+
 In this documentation you will find instructions on how to install and run the `mvt-ios` and `mvt-android` commands, and guidance on how to interpret the extracted results.
 
 ## Resources

docs/introduction.md

@@ -12,6 +12,20 @@ Mobile Verification Toolkit (MVT) is a collection of utilities designed to facil
 
 MVT is a forensic research tool intended for technologists and investigators. Using it requires understanding the basics of forensic analysis and using command-line tools. MVT is not intended for end-user self-assessment. If you are concerned with the security of your device please seek expert assistance.
 
+## Indicators of Compromise
+
+MVT supports using [indicators of compromise (IOCs)](https://github.com/mvt-project/mvt-indicators) to scan mobile devices for potential traces of targeting or infection by known spyware campaigns. This includes IOCs published by [Amnesty International](https://github.com/AmnestyTech/investigations/) and other research groups.
+
+!!! warning
+    Public indicators of compromise are insufficient to determine that a device is "clean", and not targeted with a particular spyware tool. Reliance on public indicators alone can miss recent forensic traces and give a false sense of security.
+
+    Reliable and comprehensive digital forensic support and triage requires access to non-public indicators, research and threat intelligence.
+
+    Such support is available to civil society through [Amnesty International's Security Lab](https://www.amnesty.org/en/tech/) or [Access Now’s Digital Security Helpline](https://www.accessnow.org/help/).
+
+More information about using indicators of compromise with MVT is available in the [documentation](iocs.md).
+
+
 ## Consensual Forensics
 
 While MVT is capable of extracting and processing various types of very personal records typically found on a mobile phone (such as calls history, SMS and WhatsApp messages, etc.), this is intended to help identify potential attack vectors such as malicious SMS messages leading to exploitation.

docs/iocs.md

@@ -43,6 +43,6 @@ export MVT_STIX2="/home/user/IOC1.stix2:/home/user/IOC2.stix2"
 - [This repository](https://github.com/Te-k/stalkerware-indicators) contains IOCs for Android stalkerware including [a STIX MVT-compatible file](https://raw.githubusercontent.com/Te-k/stalkerware-indicators/master/generated/stalkerware.stix2).
 - We are also maintaining [a list of IOCs](https://github.com/mvt-project/mvt-indicators) in STIX format from public spyware campaigns.
 
-You can automaticallly download the latest public indicator files with the command `mvt-ios download-iocs` or `mvt-android download-iocs`. These commands download the list of indicators listed [here](https://github.com/mvt-project/mvt/blob/main/public_indicators.json) and store them in the [appdir](https://pypi.org/project/appdirs/) folder. They are then loaded automatically by MVT.
+You can automaticallly download the latest public indicator files with the command `mvt-ios download-iocs` or `mvt-android download-iocs`. These commands download the list of indicators from the [mvt-indicators](https://github.com/mvt-project/mvt-indicators/blob/main/indicators.yaml) repository and store them in the [appdir](https://pypi.org/project/appdirs/) folder. They are then loaded automatically by MVT.
 
 Please [open an issue](https://github.com/mvt-project/mvt/issues/) to suggest new sources of STIX-formatted IOCs.

docs/ios/records.md

@@ -142,6 +142,16 @@ If indicators are provided through the command-line, they are checked against th
 
 ---
 
+### `global_preferences.json`
+
+!!! info "Availability"
+    Backup: :material-check:
+    Full filesystem dump: :material-check:
+
+This JSON file is created by mvt-ios' `GlobalPreferences` module. The module extracts records from a Plist file located at */private/var/mobile/Library/Preferences/.GlobalPreferences.plist*, which contains a system preferences including if Lockdown Mode is enabled.
+
+---
+
 ### `id_status_cache.json`
 
 !!! info "Availability"

mkdocs.yml

@@ -1,7 +1,7 @@
 site_name: Mobile Verification Toolkit
 repo_url: https://github.com/mvt-project/mvt
 edit_uri: edit/main/docs/
-copyright: Copyright © 2021-2022 MVT Project Developers
+copyright: Copyright © 2021-2023 MVT Project Developers
 site_description: Mobile Verification Toolkit Documentation
 markdown_extensions:
     - attr_list
@@ -46,4 +46,5 @@ nav:
         - Check an Android Backup (SMS messages): "android/backup.md"
         - Download APKs: "android/download_apks.md"
     - Indicators of Compromise: "iocs.md"
+    - Development: "development.md"
     - License: "license.md"

mvt/android/artifacts/__init__.py

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

mvt/android/artifacts/artifact.py

@@ -0,0 +1,36 @@
+# 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):
+    @staticmethod
+    def extract_dumpsys_section(dumpsys: str, separator: str) -> str:
+        """
+        Extract a section from a full dumpsys file.
+
+        :param dumpsys: content of the full dumpsys file (string)
+        :param separator: content of the first line separator (string)
+        :return: section extracted (string)
+        """
+        lines = []
+        in_section = False
+        for line in dumpsys.splitlines():
+            if line.strip() == separator:
+                in_section = True
+                continue
+
+            if not in_section:
+                continue
+
+            if line.strip().startswith(
+                "------------------------------------------------------------------------------"
+            ):
+                break
+
+            lines.append(line)
+
+        return "\n".join(lines)

mvt/android/artifacts/dumpsys_accessibility.py

@@ -0,0 +1,47 @@
+# 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 DumpsysAccessibilityArtifact(AndroidArtifact):
+    def check_indicators(self) -> None:
+        if not self.indicators:
+            return
+
+        for result in self.results:
+            ioc = self.indicators.check_app_id(result["package_name"])
+            if ioc:
+                result["matched_indicator"] = ioc
+                self.detected.append(result)
+                continue
+
+    def parse(self, content: str) -> None:
+        """
+        Parse the Dumpsys Accessibility section/
+        Adds results to self.results (List[Dict[str, str]])
+
+        :param content: content of the accessibility section (string)
+        """
+        in_services = False
+        for line in content.splitlines():
+            if line.strip().startswith("installed services:"):
+                in_services = True
+                continue
+
+            if not in_services:
+                continue
+
+            if line.strip() == "}":
+                break
+
+            service = line.split(":")[1].strip()
+
+            self.results.append(
+                {
+                    "package_name": service.split("/")[0],
+                    "service": service,
+                }
+            )

mvt/android/artifacts/dumpsys_appops.py

@@ -0,0 +1,150 @@
+# 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
+from typing import Any, Dict, List, Union
+
+from mvt.common.utils import convert_datetime_to_iso
+
+from .artifact import AndroidArtifact
+
+
+class DumpsysAppopsArtifact(AndroidArtifact):
+    """
+    Parser for dumpsys app ops info
+    """
+
+    def serialize(self, record: dict) -> Union[dict, list]:
+        records = []
+        for perm in record["permissions"]:
+            if "entries" not in perm:
+                continue
+
+            for entry in perm["entries"]:
+                if "timestamp" in entry:
+                    records.append(
+                        {
+                            "timestamp": entry["timestamp"],
+                            "module": self.__class__.__name__,
+                            "event": entry["access"],
+                            "data": f"{record['package_name']} access to "
+                            f"{perm['name']}: {entry['access']}",
+                        }
+                    )
+
+        return records
+
+    def check_indicators(self) -> None:
+        for result in self.results:
+            if self.indicators:
+                ioc = self.indicators.check_app_id(result.get("package_name"))
+                if ioc:
+                    result["matched_indicator"] = ioc
+                    self.detected.append(result)
+                    continue
+
+            for perm in result["permissions"]:
+                if (
+                    perm["name"] == "REQUEST_INSTALL_PACKAGES"
+                    and perm["access"] == "allow"
+                ):
+                    self.log.info(
+                        "Package %s with REQUEST_INSTALL_PACKAGES " "permission",
+                        result["package_name"],
+                    )
+
+    def parse(self, output: str) -> None:
+        self.results: List[Dict[str, Any]] = []
+        perm = {}
+        package = {}
+        entry = {}
+        uid = None
+        in_packages = False
+
+        for line in output.splitlines():
+            if line.startswith("  Uid 0:"):
+                in_packages = True
+
+            if not in_packages:
+                continue
+
+            if line.startswith("  Uid "):
+                uid = line[6:-1]
+                if entry:
+                    perm["entries"].append(entry)
+                    entry = {}
+                if package:
+                    if perm:
+                        package["permissions"].append(perm)
+
+                    perm = {}
+                    self.results.append(package)
+                package = {}
+                continue
+
+            if line.startswith("    Package "):
+                if entry:
+                    perm["entries"].append(entry)
+                    entry = {}
+
+                if package:
+                    if perm:
+                        package["permissions"].append(perm)
+
+                    perm = {}
+                    self.results.append(package)
+
+                package = {
+                    "package_name": line[12:-1],
+                    "permissions": [],
+                    "uid": uid,
+                }
+                continue
+
+            if package and line.startswith("      ") and line[6] != " ":
+                if entry:
+                    perm["entries"].append(entry)
+                    entry = {}
+                if perm:
+                    package["permissions"].append(perm)
+                    perm = {}
+
+                perm["name"] = line.split()[0]
+                perm["entries"] = []
+                if len(line.split()) > 1:
+                    perm["access"] = line.split()[1][1:-2]
+
+                continue
+
+            if line.startswith("          "):
+                # Permission entry like:
+                # Reject: [fg-s]2021-05-19 22:02:52.054 (-314d1h25m2s33ms)
+                if entry:
+                    perm["entries"].append(entry)
+                    entry = {}
+
+                entry["access"] = line.split(":")[0].strip()
+                entry["type"] = line[line.find("[") + 1 : line.find("]")]
+
+                try:
+                    entry["timestamp"] = convert_datetime_to_iso(
+                        datetime.strptime(
+                            line[line.find("]") + 1 : line.find("(")].strip(),
+                            "%Y-%m-%d %H:%M:%S.%f",
+                        )
+                    )
+                except ValueError:
+                    # Invalid date format
+                    pass
+
+            if line.strip() == "":
+                break
+
+        if entry:
+            perm["entries"].append(entry)
+        if perm:
+            package["permissions"].append(perm)
+        if package:
+            self.results.append(package)

mvt/android/artifacts/dumpsys_battery_daily.py

@@ -0,0 +1,78 @@
+# 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 typing import Union
+
+from .artifact import AndroidArtifact
+
+
+class DumpsysBatteryDailyArtifact(AndroidArtifact):
+    """
+    Parser for dumpsys dattery daily updates.
+    """
+
+    def serialize(self, record: dict) -> Union[dict, list]:
+        return {
+            "timestamp": record["from"],
+            "module": self.__class__.__name__,
+            "event": "battery_daily",
+            "data": f"Recorded update of package {record['package_name']} "
+            f"with vers {record['vers']}",
+        }
+
+    def check_indicators(self) -> None:
+        if not self.indicators:
+            return
+
+        for result in self.results:
+            ioc = self.indicators.check_app_id(result["package_name"])
+            if ioc:
+                result["matched_indicator"] = ioc
+                self.detected.append(result)
+                continue
+
+    def parse(self, output: str) -> None:
+        daily = None
+        daily_updates = []
+        for line in output.splitlines():
+            if line.startswith("  Daily from "):
+                if len(daily_updates) > 0:
+                    self.results.extend(daily_updates)
+                    daily_updates = []
+
+                timeframe = line[13:].strip()
+                date_from, date_to = timeframe.strip(":").split(" to ", 1)
+                daily = {"from": date_from[0:10], "to": date_to[0:10]}
+                continue
+
+            if not daily:
+                continue
+
+            if not line.strip().startswith("Update "):
+                continue
+
+            line = line.strip().replace("Update ", "")
+            package_name, vers = line.split(" ", 1)
+            vers_nr = vers.split("=", 1)[1]
+
+            already_seen = False
+            for update in daily_updates:
+                if package_name == update["package_name"] and vers_nr == update["vers"]:
+                    already_seen = True
+                    break
+
+            if not already_seen:
+                daily_updates.append(
+                    {
+                        "action": "update",
+                        "from": daily["from"],
+                        "to": daily["to"],
+                        "package_name": package_name,
+                        "vers": vers_nr,
+                    }
+                )
+
+        if len(daily_updates) > 0:
+            self.results.extend(daily_updates)

mvt/android/artifacts/dumpsys_battery_history.py

@@ -0,0 +1,78 @@
+# 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 DumpsysBatteryHistoryArtifact(AndroidArtifact):
+    """
+    Parser for dumpsys dattery history events.
+    """
+
+    def check_indicators(self) -> None:
+        if not self.indicators:
+            return
+
+        for result in self.results:
+            ioc = self.indicators.check_app_id(result["package_name"])
+            if ioc:
+                result["matched_indicator"] = ioc
+                self.detected.append(result)
+                continue
+
+    def parse(self, data: str) -> None:
+        for line in data.splitlines():
+            if line.startswith("Battery History "):
+                continue
+
+            if line.strip() == "":
+                break
+
+            time_elapsed = line.strip().split(" ", 1)[0]
+
+            event = ""
+            if line.find("+job") > 0:
+                event = "start_job"
+                uid = line[line.find("+job") + 5 : line.find(":")]
+                service = line[line.find(":") + 1 :].strip('"')
+                package_name = service.split("/")[0]
+            elif line.find("-job") > 0:
+                event = "end_job"
+                uid = line[line.find("-job") + 5 : line.find(":")]
+                service = line[line.find(":") + 1 :].strip('"')
+                package_name = service.split("/")[0]
+            elif line.find("+running +wake_lock=") > 0:
+                uid = line[line.find("+running +wake_lock=") + 21 : line.find(":")]
+                event = "wake"
+                service = (
+                    line[line.find("*walarm*:") + 9 :].split(" ")[0].strip('"').strip()
+                )
+                if service == "" or "/" not in service:
+                    continue
+
+                package_name = service.split("/")[0]
+            elif (line.find("+top=") > 0) or (line.find("-top") > 0):
+                if line.find("+top=") > 0:
+                    event = "start_top"
+                    top_pos = line.find("+top=")
+                else:
+                    event = "end_top"
+                    top_pos = line.find("-top=")
+                colon_pos = top_pos + line[top_pos:].find(":")
+                uid = line[top_pos + 5 : colon_pos]
+                service = ""
+                package_name = line[colon_pos + 1 :].strip('"')
+            else:
+                continue
+
+            self.results.append(
+                {
+                    "time_elapsed": time_elapsed,
+                    "event": event,
+                    "uid": uid,
+                    "package_name": package_name,
+                    "service": service,
+                }
+            )

mvt/android/artifacts/dumpsys_dbinfo.py

@@ -0,0 +1,83 @@
+# 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 .artifact import AndroidArtifact
+
+
+class DumpsysDBInfoArtifact(AndroidArtifact):
+    """
+    Parser for dumpsys DBInfo service
+    """
+
+    def check_indicators(self) -> None:
+        if not self.indicators:
+            return
+
+        for result in self.results:
+            path = result.get("path", "")
+            for part in path.split("/"):
+                ioc = self.indicators.check_app_id(part)
+                if ioc:
+                    result["matched_indicator"] = ioc
+                    self.detected.append(result)
+                    continue
+
+    def parse(self, output: str) -> None:
+        rxp = re.compile(
+            r".*\[([0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3})\].*\[Pid:\((\d+)\)\](\w+).*sql\=\"(.+?)\""
+        )  # pylint: disable=line-too-long
+        rxp_no_pid = re.compile(
+            r".*\[([0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3})\][ ]{1}(\w+).*sql\=\"(.+?)\""
+        )  # pylint: disable=line-too-long
+
+        pool = None
+        in_operations = False
+        for line in output.splitlines():
+            if line.startswith("Connection pool for "):
+                pool = line.replace("Connection pool for ", "").rstrip(":")
+
+            if not pool:
+                continue
+
+            if line.strip() == "Most recently executed operations:":
+                in_operations = True
+                continue
+
+            if not in_operations:
+                continue
+
+            if not line.startswith("        "):
+                in_operations = False
+                pool = None
+                continue
+
+            matches = rxp.findall(line)
+            if not matches:
+                matches = rxp_no_pid.findall(line)
+                if not matches:
+                    continue
+
+                match = matches[0]
+                self.results.append(
+                    {
+                        "isodate": match[0],
+                        "action": match[1],
+                        "sql": match[2],
+                        "path": pool,
+                    }
+                )
+            else:
+                match = matches[0]
+                self.results.append(
+                    {
+                        "isodate": match[0],
+                        "pid": match[1],
+                        "action": match[2],
+                        "sql": match[3],
+                        "path": pool,
+                    }
+                )

mvt/android/artifacts/dumpsys_package_activities.py

@@ -0,0 +1,84 @@
+# 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 DumpsysPackageActivitiesArtifact(AndroidArtifact):
+    def check_indicators(self) -> None:
+        if not self.indicators:
+            return
+
+        for activity in self.results:
+            ioc = self.indicators.check_app_id(activity["package_name"])
+            if ioc:
+                activity["matched_indicator"] = ioc
+                self.detected.append(activity)
+                continue
+
+    def parse(self, content: str):
+        """
+        Parse the Dumpsys Package section for activities
+        Adds results to self.results
+
+        :param content: content of the package section (string)
+        """
+        self.results = []
+
+        in_activity_resolver_table = False
+        in_non_data_actions = False
+        intent = None
+        for line in content.splitlines():
+            if line.startswith("Activity Resolver Table:"):
+                in_activity_resolver_table = True
+                continue
+
+            if not in_activity_resolver_table:
+                continue
+
+            if line.startswith("  Non-Data Actions:"):
+                in_non_data_actions = True
+                continue
+
+            if not in_non_data_actions:
+                continue
+
+            # If we hit an empty line, the Non-Data Actions section should be
+            # finished.
+            if line.strip() == "":
+                break
+
+            # We detect the action name.
+            if (
+                line.startswith(" " * 6)
+                and not line.startswith(" " * 8)
+                and ":" in line
+            ):
+                intent = line.strip().replace(":", "")
+                continue
+
+            # If we are not in an intent block yet, skip.
+            if not intent:
+                continue
+
+            # If we are in a block but the line does not start with 8 spaces
+            # it means the block ended a new one started, so we reset and
+            # continue.
+            if not line.startswith(" " * 8):
+                intent = None
+                continue
+
+            # If we got this far, we are processing receivers for the
+            # activities we are interested in.
+            activity = line.strip().split(" ")[1]
+            package_name = activity.split("/")[0]
+
+            self.results.append(
+                {
+                    "intent": intent,
+                    "package_name": package_name,
+                    "activity": activity,
+                }
+            )

mvt/android/artifacts/dumpsys_receivers.py

@@ -0,0 +1,116 @@
+# 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
+
+INTENT_NEW_OUTGOING_SMS = "android.provider.Telephony.NEW_OUTGOING_SMS"
+INTENT_SMS_RECEIVED = "android.provider.Telephony.SMS_RECEIVED"
+INTENT_DATA_SMS_RECEIVED = "android.intent.action.DATA_SMS_RECEIVED"
+INTENT_PHONE_STATE = "android.intent.action.PHONE_STATE"
+INTENT_NEW_OUTGOING_CALL = "android.intent.action.NEW_OUTGOING_CALL"
+
+
+class DumpsysReceiversArtifact(AndroidArtifact):
+    """
+    Parser for dumpsys receivers in the package section
+    """
+
+    def check_indicators(self) -> None:
+        for intent, receivers in self.results.items():
+            for receiver in receivers:
+                if intent == INTENT_NEW_OUTGOING_SMS:
+                    self.log.info(
+                        'Found a receiver to intercept outgoing SMS messages: "%s"',
+                        receiver["receiver"],
+                    )
+                elif intent == INTENT_SMS_RECEIVED:
+                    self.log.info(
+                        'Found a receiver to intercept incoming SMS messages: "%s"',
+                        receiver["receiver"],
+                    )
+                elif intent == INTENT_DATA_SMS_RECEIVED:
+                    self.log.info(
+                        'Found a receiver to intercept incoming data SMS message: "%s"',
+                        receiver["receiver"],
+                    )
+                elif intent == INTENT_PHONE_STATE:
+                    self.log.info(
+                        "Found a receiver monitoring "
+                        'telephony state/incoming calls: "%s"',
+                        receiver["receiver"],
+                    )
+                elif intent == INTENT_NEW_OUTGOING_CALL:
+                    self.log.info(
+                        'Found a receiver monitoring outgoing calls: "%s"',
+                        receiver["receiver"],
+                    )
+
+                if not self.indicators:
+                    continue
+
+                ioc = self.indicators.check_app_id(receiver["package_name"])
+                if ioc:
+                    receiver["matched_indicator"] = ioc
+                    self.detected.append({intent: receiver})
+                    continue
+
+    def parse(self, output: str) -> None:
+        self.results = {}
+
+        in_receiver_resolver_table = False
+        in_non_data_actions = False
+        intent = None
+        for line in output.splitlines():
+            if line.startswith("Receiver Resolver Table:"):
+                in_receiver_resolver_table = True
+                continue
+
+            if not in_receiver_resolver_table:
+                continue
+
+            if line.startswith("  Non-Data Actions:"):
+                in_non_data_actions = True
+                continue
+
+            if not in_non_data_actions:
+                continue
+
+            # If we hit an empty line, the Non-Data Actions section should be
+            # finished.
+            if line.strip() == "":
+                break
+
+            # We detect the action name.
+            if (
+                line.startswith(" " * 6)
+                and not line.startswith(" " * 8)
+                and ":" in line
+            ):
+                intent = line.strip().replace(":", "")
+                self.results[intent] = []
+                continue
+
+            # If we are not in an intent block yet, skip.
+            if not intent:
+                continue
+
+            # If we are in a block but the line does not start with 8 spaces
+            # it means the block ended a new one started, so we reset and
+            # continue.
+            if not line.startswith(" " * 8):
+                intent = None
+                continue
+
+            # If we got this far, we are processing receivers for the
+            # activities we are interested in.
+            receiver = line.strip().split(" ")[1]
+            package_name = receiver.split("/")[0]
+
+            self.results[intent].append(
+                {
+                    "package_name": package_name,
+                    "receiver": receiver,
+                }
+            )

mvt/android/artifacts/getprop.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 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,70 @@
+# 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,72 @@
+# 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

@@ -8,10 +8,18 @@ import logging
 import click
 
 from mvt.common.cmd_check_iocs import CmdCheckIOCS
-from mvt.common.help import (HELP_MSG_FAST, HELP_MSG_HASHES, HELP_MSG_IOC,
-                             HELP_MSG_LIST_MODULES, HELP_MSG_MODULE,
-                             HELP_MSG_OUTPUT, HELP_MSG_SERIAL,
-                             HELP_MSG_VERBOSE)
+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,
+)
 from mvt.common.logo import logo
 from mvt.common.updates import IndicatorsUpdates
 from mvt.common.utils import init_logging, set_verbose_logging
@@ -24,43 +32,60 @@ 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'])
+
+CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])
 
 
-#==============================================================================
+# ==============================================================================
 # Main
-#==============================================================================
+# ==============================================================================
 @click.group(invoke_without_command=False)
 def cli():
     logo()
 
 
-#==============================================================================
+# ==============================================================================
 # Command: version
-#==============================================================================
+# ==============================================================================
 @cli.command("version", help="Show the currently installed version of MVT")
 def version():
     return
 
 
-#==============================================================================
+# ==============================================================================
 # Command: download-apks
-#==============================================================================
-@cli.command("download-apks", help="Download all or only non-system installed APKs",
-             context_settings=CONTEXT_SETTINGS)
+# ==============================================================================
+@cli.command(
+    "download-apks",
+    help="Download all or only non-system installed APKs",
+    context_settings=CONTEXT_SETTINGS,
+)
 @click.option("--serial", "-s", type=str, help=HELP_MSG_SERIAL)
-@click.option("--all-apks", "-a", is_flag=True,
-              help="Extract all packages installed on the phone, including system packages")
+@click.option(
+    "--all-apks",
+    "-a",
+    is_flag=True,
+    help="Extract all packages installed on the phone, including system packages",
+)
 @click.option("--virustotal", "-v", is_flag=True, help="Check packages on VirusTotal")
-@click.option("--output", "-o", type=click.Path(exists=False),
-              help="Specify a path to a folder where you want to store the APKs")
-@click.option("--from-file", "-f", type=click.Path(exists=True),
-              help="Instead of acquiring from phone, load an existing packages.json file for "
-                   "lookups (mainly for debug purposes)")
+@click.option(
+    "--output",
+    "-o",
+    type=click.Path(exists=False),
+    help="Specify a path to a folder where you want to store the APKs",
+)
+@click.option(
+    "--from-file",
+    "-f",
+    type=click.Path(exists=True),
+    help="Instead of acquiring from phone, load an existing packages.json file for "
+    "lookups (mainly for debug purposes)",
+)
 @click.option("--verbose", "-v", is_flag=True, help=HELP_MSG_VERBOSE)
 @click.pass_context
 def download_apks(ctx, all_apks, virustotal, output, from_file, serial, verbose):
@@ -99,25 +124,57 @@ def download_apks(ctx, all_apks, virustotal, output, from_file, serial, verbose)
         ctx.exit(1)
 
 
-#==============================================================================
+# ==============================================================================
 # Command: check-adb
-#==============================================================================
-@cli.command("check-adb", help="Check an Android device over adb",
-             context_settings=CONTEXT_SETTINGS)
+# ==============================================================================
+@cli.command(
+    "check-adb",
+    help="Check an Android device over ADB",
+    context_settings=CONTEXT_SETTINGS,
+)
 @click.option("--serial", "-s", type=str, help=HELP_MSG_SERIAL)
-@click.option("--iocs", "-i", type=click.Path(exists=True), multiple=True,
-              default=[], help=HELP_MSG_IOC)
-@click.option("--output", "-o", type=click.Path(exists=False),
-              help=HELP_MSG_OUTPUT)
+@click.option(
+    "--iocs",
+    "-i",
+    type=click.Path(exists=True),
+    multiple=True,
+    default=[],
+    help=HELP_MSG_IOC,
+)
+@click.option("--output", "-o", type=click.Path(exists=False), help=HELP_MSG_OUTPUT)
 @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)
-    cmd = CmdAndroidCheckADB(results_path=output, ioc_files=iocs,
-                             module_name=module, serial=serial, 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,
+        ioc_files=iocs,
+        module_name=module,
+        serial=serial,
+        module_options=module_options,
+    )
 
     if list_modules:
         cmd.list_modules()
@@ -128,19 +185,29 @@ def check_adb(ctx, serial, iocs, output, fast, list_modules, module, verbose):
     cmd.run()
 
     if cmd.detected_count > 0:
-        log.warning("The analysis of the Android device produced %d detections!",
-                    cmd.detected_count)
+        log.warning(
+            "The analysis of the Android device produced %d detections!",
+            cmd.detected_count,
+        )
 
 
-#==============================================================================
+# ==============================================================================
 # Command: check-bugreport
-#==============================================================================
-@cli.command("check-bugreport", help="Check an Android Bug Report",
-             context_settings=CONTEXT_SETTINGS)
-@click.option("--iocs", "-i", type=click.Path(exists=True), multiple=True,
-              default=[], help=HELP_MSG_IOC)
-@click.option("--output", "-o", type=click.Path(exists=False),
-              help=HELP_MSG_OUTPUT)
+# ==============================================================================
+@cli.command(
+    "check-bugreport",
+    help="Check an Android Bug Report",
+    context_settings=CONTEXT_SETTINGS,
+)
+@click.option(
+    "--iocs",
+    "-i",
+    type=click.Path(exists=True),
+    multiple=True,
+    default=[],
+    help=HELP_MSG_IOC,
+)
+@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("--module", "-m", help=HELP_MSG_MODULE)
 @click.option("--verbose", "-v", is_flag=True, help=HELP_MSG_VERBOSE)
@@ -148,10 +215,14 @@ def check_adb(ctx, serial, iocs, output, fast, list_modules, module, verbose):
 @click.pass_context
 def check_bugreport(ctx, iocs, output, list_modules, module, verbose, bugreport_path):
     set_verbose_logging(verbose)
-    # Always generate hashes as bug reports are small.
-    cmd = CmdAndroidCheckBugreport(target_path=bugreport_path,
-                                   results_path=output, ioc_files=iocs,
-                                   module_name=module, hashes=True)
+    # Always generate hashes as bug reports are small.
+    cmd = CmdAndroidCheckBugreport(
+        target_path=bugreport_path,
+        results_path=output,
+        ioc_files=iocs,
+        module_name=module,
+        hashes=True,
+    )
 
     if list_modules:
         cmd.list_modules()
@@ -162,28 +233,56 @@ def check_bugreport(ctx, iocs, output, list_modules, module, verbose, bugreport_
     cmd.run()
 
     if cmd.detected_count > 0:
-        log.warning("The analysis of the Android bug report produced %d detections!",
-                    cmd.detected_count)
+        log.warning(
+            "The analysis of the Android bug report produced %d detections!",
+            cmd.detected_count,
+        )
 
 
-#==============================================================================
+# ==============================================================================
 # Command: check-backup
-#==============================================================================
-@cli.command("check-backup", help="Check an Android Backup",
-             context_settings=CONTEXT_SETTINGS)
-@click.option("--iocs", "-i", type=click.Path(exists=True), multiple=True,
-              default=[], help=HELP_MSG_IOC)
-@click.option("--output", "-o", type=click.Path(exists=False),
-              help=HELP_MSG_OUTPUT)
+# ==============================================================================
+@cli.command(
+    "check-backup", help="Check an Android Backup", context_settings=CONTEXT_SETTINGS
+)
+@click.option(
+    "--iocs",
+    "-i",
+    type=click.Path(exists=True),
+    multiple=True,
+    default=[],
+    help=HELP_MSG_IOC,
+)
+@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)
+    cmd = CmdAndroidCheckBackup(
+        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:
         cmd.list_modules()
@@ -194,30 +293,62 @@ def check_backup(ctx, iocs, output, list_modules, verbose, backup_path):
     cmd.run()
 
     if cmd.detected_count > 0:
-        log.warning("The analysis of the Android backup produced %d detections!",
-                    cmd.detected_count)
+        log.warning(
+            "The analysis of the Android backup produced %d detections!",
+            cmd.detected_count,
+        )
 
 
-#==============================================================================
+# ==============================================================================
 # Command: check-androidqf
-#==============================================================================
-@cli.command("check-androidqf", help="Check data collected with AndroidQF",
-             context_settings=CONTEXT_SETTINGS)
-@click.option("--iocs", "-i", type=click.Path(exists=True), multiple=True,
-              default=[], help=HELP_MSG_IOC)
-@click.option("--output", "-o", type=click.Path(exists=False),
-              help=HELP_MSG_OUTPUT)
+# ==============================================================================
+@cli.command(
+    "check-androidqf",
+    help="Check data collected with AndroidQF",
+    context_settings=CONTEXT_SETTINGS,
+)
+@click.option(
+    "--iocs",
+    "-i",
+    type=click.Path(exists=True),
+    multiple=True,
+    default=[],
+    help=HELP_MSG_IOC,
+)
+@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("--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):
+def check_androidqf(
+    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)
+
+    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:
         cmd.list_modules()
@@ -228,17 +359,28 @@ def check_androidqf(ctx, iocs, output, list_modules, module, hashes, verbose, an
     cmd.run()
 
     if cmd.detected_count > 0:
-        log.warning("The analysis of the AndroidQF acquisition produced %d detections!",
-                    cmd.detected_count)
+        log.warning(
+            "The analysis of the AndroidQF acquisition produced %d detections!",
+            cmd.detected_count,
+        )
 
 
-#==============================================================================
+# ==============================================================================
 # Command: check-iocs
-#==============================================================================
-@cli.command("check-iocs", help="Compare stored JSON results to provided indicators",
-             context_settings=CONTEXT_SETTINGS)
-@click.option("--iocs", "-i", type=click.Path(exists=True), multiple=True,
-              default=[], help=HELP_MSG_IOC)
+# ==============================================================================
+@cli.command(
+    "check-iocs",
+    help="Compare stored JSON results to provided indicators",
+    context_settings=CONTEXT_SETTINGS,
+)
+@click.option(
+    "--iocs",
+    "-i",
+    type=click.Path(exists=True),
+    multiple=True,
+    default=[],
+    help=HELP_MSG_IOC,
+)
 @click.option("--list-modules", "-l", is_flag=True, help=HELP_MSG_LIST_MODULES)
 @click.option("--module", "-m", help=HELP_MSG_MODULE)
 @click.argument("FOLDER", type=click.Path(exists=True))
@@ -254,11 +396,14 @@ def check_iocs(ctx, iocs, list_modules, module, folder):
     cmd.run()
 
 
-#==============================================================================
+# ==============================================================================
 # Command: download-iocs
-#==============================================================================
-@cli.command("download-iocs", help="Download public STIX2 indicators",
-             context_settings=CONTEXT_SETTINGS)
+# ==============================================================================
+@cli.command(
+    "download-iocs",
+    help="Download public STIX2 indicators",
+    context_settings=CONTEXT_SETTINGS,
+)
 def download_indicators():
     ioc_updates = IndicatorsUpdates()
     ioc_updates.update()

mvt/android/cmd_check_adb.py

@@ -14,7 +14,6 @@ log = logging.getLogger(__name__)
 
 
 class CmdAndroidCheckADB(Command):
-
     def __init__(
         self,
         target_path: Optional[str] = None,
@@ -22,11 +21,17 @@ class CmdAndroidCheckADB(Command):
         ioc_files: Optional[list] = None,
         module_name: Optional[str] = None,
         serial: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
     ) -> None:
-        super().__init__(target_path=target_path, results_path=results_path,
-                         ioc_files=ioc_files, module_name=module_name,
-                         serial=serial, fast_mode=fast_mode, log=log)
+        super().__init__(
+            target_path=target_path,
+            results_path=results_path,
+            ioc_files=ioc_files,
+            module_name=module_name,
+            serial=serial,
+            module_options=module_options,
+            log=log,
+        )
 
         self.name = "check-adb"
         self.modules = ADB_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
 
@@ -14,7 +17,6 @@ log = logging.getLogger(__name__)
 
 
 class CmdAndroidCheckAndroidQF(Command):
-
     def __init__(
         self,
         target_path: Optional[str] = None,
@@ -22,13 +24,44 @@ class CmdAndroidCheckAndroidQF(Command):
         ioc_files: Optional[list] = None,
         module_name: Optional[str] = None,
         serial: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
-        hashes: Optional[bool] = False,
+        module_options: Optional[dict] = None,
+        hashes: bool = False,
     ) -> None:
-        super().__init__(target_path=target_path, results_path=results_path,
-                         ioc_files=ioc_files, module_name=module_name,
-                         serial=serial, fast_mode=fast_mode, hashes=hashes,
-                         log=log)
+        super().__init__(
+            target_path=target_path,
+            results_path=results_path,
+            ioc_files=ioc_files,
+            module_name=module_name,
+            serial=serial,
+            module_options=module_options,
+            hashes=hashes,
+            log=log,
+        )
 
         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,12 +11,14 @@ 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,
-                                        InvalidBackupPassword, parse_ab_header,
-                                        parse_backup_file)
+from mvt.android.modules.backup.helpers import prompt_or_load_android_backup_password
+from mvt.android.parsers.backup import (
+    AndroidBackupParsingError,
+    InvalidBackupPassword,
+    parse_ab_header,
+    parse_backup_file,
+)
 from mvt.common.command import Command
 
 from .modules.backup import BACKUP_MODULES
@@ -25,7 +27,6 @@ log = logging.getLogger(__name__)
 
 
 class CmdAndroidCheckBackup(Command):
-
     def __init__(
         self,
         target_path: Optional[str] = None,
@@ -33,13 +34,19 @@ class CmdAndroidCheckBackup(Command):
         ioc_files: Optional[list] = None,
         module_name: Optional[str] = None,
         serial: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
-        hashes: Optional[bool] = False,
+        module_options: Optional[dict] = None,
+        hashes: bool = False,
     ) -> None:
-        super().__init__(target_path=target_path, results_path=results_path,
-                         ioc_files=ioc_files, module_name=module_name,
-                         serial=serial, fast_mode=fast_mode, hashes=hashes,
-                         log=log)
+        super().__init__(
+            target_path=target_path,
+            results_path=results_path,
+            ioc_files=ioc_files,
+            module_name=module_name,
+            serial=serial,
+            module_options=module_options,
+            hashes=hashes,
+            log=log,
+        )
 
         self.name = "check-backup"
         self.modules = BACKUP_MODULES
@@ -64,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:
@@ -85,16 +97,18 @@ class CmdAndroidCheckBackup(Command):
             self.target_path = Path(self.target_path).absolute().as_posix()
             for root, subdirs, subfiles in os.walk(os.path.abspath(self.target_path)):
                 for fname in subfiles:
-                    self.backup_files.append(os.path.relpath(os.path.join(root, fname),
-                                                             self.target_path))
+                    self.backup_files.append(
+                        os.path.relpath(os.path.join(root, fname), self.target_path)
+                    )
         else:
-            log.critical("Invalid backup path, path should be a folder or an "
-                         "Android Backup (.ab) file")
+            log.critical(
+                "Invalid backup path, path should be a folder or an "
+                "Android Backup (.ab) file"
+            )
             sys.exit(1)
 
     def module_init(self, module: BackupExtraction) -> None:  # type: ignore[override]
         if self.backup_type == "folder":
             module.from_folder(self.target_path, self.backup_files)
         else:
-            module.from_ab(self.target_path, self.backup_archive,
-                           self.backup_files)
+            module.from_ab(self.target_path, self.backup_archive, self.backup_files)

mvt/android/cmd_check_bugreport.py

@@ -18,7 +18,6 @@ log = logging.getLogger(__name__)
 
 
 class CmdAndroidCheckBugreport(Command):
-
     def __init__(
         self,
         target_path: Optional[str] = None,
@@ -26,13 +25,19 @@ class CmdAndroidCheckBugreport(Command):
         ioc_files: Optional[list] = None,
         module_name: Optional[str] = None,
         serial: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
-        hashes: Optional[bool] = False,
+        module_options: Optional[dict] = None,
+        hashes: bool = False,
     ) -> None:
-        super().__init__(target_path=target_path, results_path=results_path,
-                         ioc_files=ioc_files, module_name=module_name,
-                         serial=serial, fast_mode=fast_mode, hashes=hashes,
-                         log=log)
+        super().__init__(
+            target_path=target_path,
+            results_path=results_path,
+            ioc_files=ioc_files,
+            module_name=module_name,
+            serial=serial,
+            module_options=module_options,
+            hashes=hashes,
+            log=log,
+        )
 
         self.name = "check-bugreport"
         self.modules = BUGREPORT_MODULES
@@ -55,8 +60,9 @@ class CmdAndroidCheckBugreport(Command):
             parent_path = Path(self.target_path).absolute().as_posix()
             for root, _, subfiles in os.walk(os.path.abspath(self.target_path)):
                 for file_name in subfiles:
-                    file_path = os.path.relpath(os.path.join(root, file_name),
-                                                parent_path)
+                    file_path = os.path.relpath(
+                        os.path.join(root, file_name), parent_path
+                    )
                     self.bugreport_files.append(file_path)
 
     def module_init(self, module: BugReportModule) -> None:  # type: ignore[override]

mvt/android/cmd_download_apks.py

@@ -26,7 +26,7 @@ class DownloadAPKs(AndroidExtraction):
     def __init__(
         self,
         results_path: Optional[str] = None,
-        all_apks: Optional[bool] = False,
+        all_apks: bool = False,
         packages: Optional[list] = None,
     ) -> None:
         """Initialize module.
@@ -66,27 +66,31 @@ class DownloadAPKs(AndroidExtraction):
         if "==/" in remote_path:
             file_name = "_" + remote_path.split("==/")[1].replace(".apk", "")
 
-        local_path = os.path.join(self.results_path_apks,
-                                  f"{package_name}{file_name}.apk")
+        local_path = os.path.join(
+            self.results_path_apks, f"{package_name}{file_name}.apk"
+        )
         name_counter = 0
         while True:
             if not os.path.exists(local_path):
                 break
 
             name_counter += 1
-            local_path = os.path.join(self.results_path_apks,
-                                      f"{package_name}{file_name}_{name_counter}.apk")
+            local_path = os.path.join(
+                self.results_path_apks, f"{package_name}{file_name}_{name_counter}.apk"
+            )
 
         try:
             self._adb_download(remote_path, local_path)
         except InsufficientPrivileges:
-            log.error("Unable to pull package file from %s: insufficient privileges, "
-                      "it might be a system app", remote_path)
+            log.error(
+                "Unable to pull package file from %s: insufficient privileges, "
+                "it might be a system app",
+                remote_path,
+            )
             self._adb_reconnect()
             return None
         except Exception as exc:
-            log.exception("Failed to pull package file from %s: %s",
-                          remote_path, exc)
+            log.exception("Failed to pull package file from %s: %s", remote_path, exc)
             self._adb_reconnect()
             return None
 
@@ -106,10 +110,10 @@ class DownloadAPKs(AndroidExtraction):
         self.packages = m.results
 
     def pull_packages(self) -> None:
-        """Download all files of all selected packages from the device.
-        """
-        log.info("Starting extraction of installed APKs at folder %s",
-                 self.results_path)
+        """Download all files of all selected packages from the device."""
+        log.info(
+            "Starting extraction of installed APKs at folder %s", self.results_path
+        )
 
         # If the user provided the flag --all-apks we select all packages.
         packages_selection = []
@@ -123,8 +127,10 @@ class DownloadAPKs(AndroidExtraction):
                 if not package.get("system", False):
                     packages_selection.append(package)
 
-            log.info("Selected only %d packages which are not marked as \"system\"",
-                     len(packages_selection))
+            log.info(
+                'Selected only %d packages which are not marked as "system"',
+                len(packages_selection),
+            )
 
         if len(packages_selection) == 0:
             log.info("No packages were selected for download")
@@ -136,19 +142,26 @@ class DownloadAPKs(AndroidExtraction):
         if not os.path.exists(self.results_path_apks):
             os.makedirs(self.results_path_apks, exist_ok=True)
 
-        for i in track(range(len(packages_selection)),
-                       description=f"Downloading {len(packages_selection)} packages..."):
+        for i in track(
+            range(len(packages_selection)),
+            description=f"Downloading {len(packages_selection)} packages...",
+        ):
             package = packages_selection[i]
 
-            log.info("[%d/%d] Package: %s", i, len(packages_selection),
-                     package["package_name"])
+            log.info(
+                "[%d/%d] Package: %s",
+                i,
+                len(packages_selection),
+                package["package_name"],
+            )
 
             # Sometimes the package path contains multiple lines for multiple
             # apks. We loop through each line and download each file.
             for package_file in package["files"]:
                 device_path = package_file["path"]
-                local_path = self.pull_package_file(package["package_name"],
-                                                    device_path)
+                local_path = self.pull_package_file(
+                    package["package_name"], device_path
+                )
                 if not local_path:
                     continue
 

mvt/android/modules/adb/__init__.py

@@ -23,8 +23,24 @@ from .settings import Settings
 from .sms import SMS
 from .whatsapp import Whatsapp
 
-ADB_MODULES = [ChromeHistory, SMS, Whatsapp, Processes, Getprop, Settings,
-               SELinuxStatus, DumpsysBatteryHistory, DumpsysBatteryDaily,
-               DumpsysReceivers, DumpsysActivities, DumpsysAccessibility,
-               DumpsysDBInfo, DumpsysFull, DumpsysAppOps, Packages, Logcat,
-               RootBinaries, Files]
+ADB_MODULES = [
+    ChromeHistory,
+    SMS,
+    Whatsapp,
+    Processes,
+    Getprop,
+    Settings,
+    SELinuxStatus,
+    DumpsysBatteryHistory,
+    DumpsysBatteryDaily,
+    DumpsysReceivers,
+    DumpsysActivities,
+    DumpsysAccessibility,
+    DumpsysDBInfo,
+    DumpsysFull,
+    DumpsysAppOps,
+    Packages,
+    Logcat,
+    RootBinaries,
+    Files,
+]

mvt/android/modules/adb/base.py

@@ -16,13 +16,20 @@ from typing import Callable, Optional
 from adb_shell.adb_device import AdbDeviceTcp, AdbDeviceUsb
 from adb_shell.auth.keygen import keygen, write_public_keyfile
 from adb_shell.auth.sign_pythonrsa import PythonRSASigner
-from adb_shell.exceptions import (AdbCommandFailureException, DeviceAuthError,
-                                  UsbDeviceNotFoundError, UsbReadFailedError)
-from rich.prompt import Prompt
+from adb_shell.exceptions import (
+    AdbCommandFailureException,
+    DeviceAuthError,
+    UsbDeviceNotFoundError,
+    UsbReadFailedError,
+)
 from usb1 import USBErrorAccess, USBErrorBusy
 
-from mvt.android.parsers.backup import (InvalidBackupPassword, parse_ab_header,
-                                        parse_backup_file)
+from mvt.android.modules.backup.helpers import prompt_or_load_android_backup_password
+from mvt.android.parsers.backup import (
+    InvalidBackupPassword,
+    parse_ab_header,
+    parse_backup_file,
+)
 from mvt.common.module import InsufficientPrivileges, MVTModule
 
 ADB_KEY_PATH = os.path.expanduser("~/.android/adbkey")
@@ -37,13 +44,18 @@ class AndroidExtraction(MVTModule):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
         self.device = None
         self.serial = None
@@ -78,36 +90,49 @@ class AndroidExtraction(MVTModule):
             try:
                 self.device = AdbDeviceUsb(serial=self.serial)
             except UsbDeviceNotFoundError:
-                self.log.critical("No device found. Make sure it is connected and unlocked.")
+                self.log.critical(
+                    "No device found. Make sure it is connected and unlocked."
+                )
                 sys.exit(-1)
         # Otherwise we try to use the TCP transport.
         else:
             addr = self.serial.split(":")
             if len(addr) < 2:
-                raise ValueError("TCP serial number must follow the format: `address:port`")
+                raise ValueError(
+                    "TCP serial number must follow the format: `address:port`"
+                )
 
-            self.device = AdbDeviceTcp(addr[0], int(addr[1]),
-                                       default_transport_timeout_s=30.)
+            self.device = AdbDeviceTcp(
+                addr[0], int(addr[1]), default_transport_timeout_s=30.0
+            )
 
         while True:
             try:
                 self.device.connect(rsa_keys=[signer], auth_timeout_s=5)
             except (USBErrorBusy, USBErrorAccess):
-                self.log.critical("Device is busy, maybe run `adb kill-server` and try again.")
+                self.log.critical(
+                    "Device is busy, maybe run `adb kill-server` and try again."
+                )
                 sys.exit(-1)
             except DeviceAuthError:
-                self.log.error("You need to authorize this computer on the Android device. "
-                               "Retrying in 5 seconds...")
+                self.log.error(
+                    "You need to authorize this computer on the Android device. "
+                    "Retrying in 5 seconds..."
+                )
                 time.sleep(5)
             except UsbReadFailedError:
-                self.log.error("Unable to connect to the device over USB. "
-                               "Try to unplug, plug the device and start again.")
+                self.log.error(
+                    "Unable to connect to the device over USB. "
+                    "Try to unplug, plug the device and start again."
+                )
                 sys.exit(-1)
             except OSError as exc:
                 if exc.errno == 113 and self.serial:
-                    self.log.critical("Unable to connect to the device %s: "
-                                      "did you specify the correct IP address?",
-                                      self.serial)
+                    self.log.critical(
+                        "Unable to connect to the device %s: "
+                        "did you specify the correct IP address?",
+                        self.serial,
+                    )
                     sys.exit(-1)
             else:
                 break
@@ -144,9 +169,11 @@ class AndroidExtraction(MVTModule):
     def _adb_root_or_die(self) -> None:
         """Check if we have a `su` binary, otherwise raise an Exception."""
         if not self._adb_check_if_root():
-            raise InsufficientPrivileges("This module is optionally available "
-                                         "in case the device is already rooted."
-                                         " Do NOT root your own device!")
+            raise InsufficientPrivileges(
+                "This module is optionally available "
+                "in case the device is already rooted."
+                " Do NOT root your own device!"
+            )
 
     def _adb_command_as_root(self, command):
         """Execute an adb shell command.
@@ -177,7 +204,7 @@ class AndroidExtraction(MVTModule):
         remote_path: str,
         local_path: str,
         progress_callback: Optional[Callable] = None,
-        retry_root: Optional[bool] = True
+        retry_root: Optional[bool] = True,
     ) -> None:
         """Download a file form the device.
 
@@ -192,41 +219,48 @@ class AndroidExtraction(MVTModule):
             self.device.pull(remote_path, local_path, progress_callback)
         except AdbCommandFailureException as exc:
             if retry_root:
-                self._adb_download_root(remote_path, local_path,
-                                        progress_callback)
+                self._adb_download_root(remote_path, local_path, progress_callback)
             else:
-                raise Exception(f"Unable to download file {remote_path}: {exc}") from exc
+                raise Exception(
+                    f"Unable to download file {remote_path}: {exc}"
+                ) from exc
 
     def _adb_download_root(
         self,
         remote_path: str,
         local_path: str,
-        progress_callback: Optional[Callable] = None
+        progress_callback: Optional[Callable] = None,
     ) -> None:
         try:
             # Check if we have root, if not raise an Exception.
             self._adb_root_or_die()
 
             # We generate a random temporary filename.
-            allowed_chars = (string.ascii_uppercase
-                             + string.ascii_lowercase
-                             + string.digits)
-            tmp_filename = "tmp_" + ''.join(random.choices(allowed_chars, k=10))
+            allowed_chars = (
+                string.ascii_uppercase + string.ascii_lowercase + string.digits
+            )
+            tmp_filename = "tmp_" + "".join(random.choices(allowed_chars, k=10))
 
             # We create a temporary local file.
             new_remote_path = f"/sdcard/{tmp_filename}"
 
             # We copy the file from the data folder to /sdcard/.
             cp_output = self._adb_command_as_root(f"cp {remote_path} {new_remote_path}")
-            if cp_output.startswith("cp: ") and "No such file or directory" in cp_output:
+            if (
+                cp_output.startswith("cp: ")
+                and "No such file or directory" in cp_output
+            ):
                 raise Exception(f"Unable to process file {remote_path}: File not found")
             if cp_output.startswith("cp: ") and "Permission denied" in cp_output:
-                raise Exception(f"Unable to process file {remote_path}: Permission denied")
+                raise Exception(
+                    f"Unable to process file {remote_path}: Permission denied"
+                )
 
             # We download from /sdcard/ to the local temporary file.
             # If it doesn't work now, don't try again (retry_root=False)
-            self._adb_download(new_remote_path, local_path, progress_callback,
-                               retry_root=False)
+            self._adb_download(
+                new_remote_path, local_path, progress_callback, retry_root=False
+            )
 
             # Delete the copy on /sdcard/.
             self._adb_command(f"rm -rf {new_remote_path}")
@@ -234,8 +268,7 @@ class AndroidExtraction(MVTModule):
         except AdbCommandFailureException as exc:
             raise Exception(f"Unable to download file {remote_path}: {exc}") from exc
 
-    def _adb_process_file(self, remote_path: str,
-                          process_routine: Callable) -> None:
+    def _adb_process_file(self, remote_path: str, process_routine: Callable) -> None:
         """Download a local copy of a file which is only accessible as root.
         This is a wrapper around process_routine.
 
@@ -273,8 +306,16 @@ class AndroidExtraction(MVTModule):
         self._adb_command(f"rm -f {new_remote_path}")
 
     def _generate_backup(self, package_name: str) -> bytes:
-        self.log.info("Please check phone and accept Android backup prompt. "
-                      "You may need to set a backup password. \a")
+        self.log.info(
+            "Please check phone and accept Android backup prompt. "
+            "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...
@@ -284,19 +325,24 @@ class AndroidExtraction(MVTModule):
         header = parse_ab_header(backup_output)
 
         if not header["backup"]:
-            self.log.error("Extracting SMS via Android backup failed. "
-                           "No valid backup data found.")
+            self.log.error(
+                "Extracting SMS via Android backup failed. "
+                "No valid backup data found."
+            )
             return None
 
         if header["encryption"] == "none":
             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)
+                decrypted_backup_tar = parse_backup_file(backup_output, backup_password)
                 return decrypted_backup_tar
             except InvalidBackupPassword:
                 self.log.error("You provided the wrong password! Please try again...")

mvt/android/modules/adb/chrome_history.py

@@ -8,8 +8,7 @@ import os
 import sqlite3
 from typing import Optional, Union
 
-from mvt.common.utils import (convert_chrometime_to_datetime,
-                              convert_datetime_to_iso)
+from mvt.common.utils import convert_chrometime_to_datetime, convert_datetime_to_iso
 
 from .base import AndroidExtraction
 
@@ -24,13 +23,18 @@ class ChromeHistory(AndroidExtraction):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
         self.results = []
 
     def serialize(self, record: dict) -> Union[dict, list]:
@@ -39,7 +43,7 @@ class ChromeHistory(AndroidExtraction):
             "module": self.__class__.__name__,
             "event": "visit",
             "data": f"{record['id']} - {record['url']} (visit ID: {record['visit_id']}, "
-                    f"redirect source: {record['redirect_source']})"
+            f"redirect source: {record['redirect_source']})",
         }
 
     def check_indicators(self) -> None:
@@ -59,7 +63,8 @@ class ChromeHistory(AndroidExtraction):
         assert isinstance(self.results, list)  # assert results type for mypy
         conn = sqlite3.connect(db_path)
         cur = conn.cursor()
-        cur.execute("""
+        cur.execute(
+            """
             SELECT
                 urls.id,
                 urls.url,
@@ -69,31 +74,35 @@ class ChromeHistory(AndroidExtraction):
             FROM urls
             JOIN visits ON visits.url = urls.id
             ORDER BY visits.visit_time;
-        """)
+        """
+        )
 
         for item in cur:
-            self.results.append({
-                "id": item[0],
-                "url": item[1],
-                "visit_id": item[2],
-                "timestamp": item[3],
-                "isodate": convert_datetime_to_iso(
-                    convert_chrometime_to_datetime(item[3])),
-                "redirect_source": item[4],
-            })
+            self.results.append(
+                {
+                    "id": item[0],
+                    "url": item[1],
+                    "visit_id": item[2],
+                    "timestamp": item[3],
+                    "isodate": convert_datetime_to_iso(
+                        convert_chrometime_to_datetime(item[3])
+                    ),
+                    "redirect_source": item[4],
+                }
+            )
 
         cur.close()
         conn.close()
 
-        self.log.info("Extracted a total of %d history items",
-                      len(self.results))
+        self.log.info("Extracted a total of %d history items", len(self.results))
 
     def run(self) -> None:
         self._adb_connect()
 
         try:
-            self._adb_process_file(os.path.join("/", CHROME_HISTORY_PATH),
-                                   self._parse_db)
+            self._adb_process_file(
+                os.path.join("/", CHROME_HISTORY_PATH), self._parse_db
+            )
         except Exception as exc:
             self.log.error(exc)
 

mvt/android/modules/adb/dumpsys_accessibility.py

@@ -6,12 +6,12 @@
 import logging
 from typing import Optional
 
-from mvt.android.parsers import parse_dumpsys_accessibility
+from mvt.android.artifacts.dumpsys_accessibility import DumpsysAccessibilityArtifact
 
 from .base import AndroidExtraction
 
 
-class DumpsysAccessibility(AndroidExtraction):
+class DumpsysAccessibility(DumpsysAccessibilityArtifact, AndroidExtraction):
     """This module extracts stats on accessibility."""
 
     def __init__(
@@ -19,35 +19,31 @@ class DumpsysAccessibility(AndroidExtraction):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
-
-    def check_indicators(self) -> None:
-        if not self.indicators:
-            return
-
-        for result in self.results:
-            ioc = self.indicators.check_app_id(result["package_name"])
-            if ioc:
-                result["matched_indicator"] = ioc
-                self.detected.append(result)
-                continue
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
     def run(self) -> None:
         self._adb_connect()
         output = self._adb_command("dumpsys accessibility")
         self._adb_disconnect()
 
-        self.results = parse_dumpsys_accessibility(output)
+        self.parse(output)
 
         for result in self.results:
-            self.log.info("Found installed accessibility service \"%s\"",
-                          result.get("service"))
+            self.log.info(
+                'Found installed accessibility service "%s"', result.get("service")
+            )
 
-        self.log.info("Identified a total of %d accessibility services",
-                      len(self.results))
+        self.log.info(
+            "Identified a total of %d accessibility services", len(self.results)
+        )

mvt/android/modules/adb/dumpsys_activities.py

@@ -6,12 +6,14 @@
 import logging
 from typing import Optional
 
-from mvt.android.parsers import parse_dumpsys_activity_resolver_table
+from mvt.android.artifacts.dumpsys_package_activities import (
+    DumpsysPackageActivitiesArtifact,
+)
 
 from .base import AndroidExtraction
 
 
-class DumpsysActivities(AndroidExtraction):
+class DumpsysActivities(DumpsysPackageActivitiesArtifact, AndroidExtraction):
     """This module extracts details on receivers for risky activities."""
 
     def __init__(
@@ -19,33 +21,25 @@ class DumpsysActivities(AndroidExtraction):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
-        self.results = results if results else {}
-
-    def check_indicators(self) -> None:
-        if not self.indicators:
-            return
-
-        for intent, activities in self.results.items():
-            for activity in activities:
-                ioc = self.indicators.check_app_id(activity["package_name"])
-                if ioc:
-                    activity["matched_indicator"] = ioc
-                    self.detected.append({intent: activity})
-                    continue
+        self.results = results if results else []
 
     def run(self) -> None:
         self._adb_connect()
         output = self._adb_command("dumpsys package")
         self._adb_disconnect()
+        self.parse(output)
 
-        self.results = parse_dumpsys_activity_resolver_table(output)
-
-        self.log.info("Extracted activities for %d intents", len(self.results))
+        self.log.info("Extracted %d package activities", len(self.results))

mvt/android/modules/adb/dumpsys_appops.py

@@ -4,14 +4,14 @@
 #   https://license.mvt.re/1.1/
 
 import logging
-from typing import Optional, Union
+from typing import Optional
 
-from mvt.android.parsers.dumpsys import parse_dumpsys_appops
+from mvt.android.artifacts.dumpsys_appops import DumpsysAppopsArtifact
 
 from .base import AndroidExtraction
 
 
-class DumpsysAppOps(AndroidExtraction):
+class DumpsysAppOps(DumpsysAppopsArtifact, AndroidExtraction):
     """This module extracts records from App-op Manager."""
 
     slug = "dumpsys_appops"
@@ -21,53 +21,26 @@ class DumpsysAppOps(AndroidExtraction):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
-
-    def serialize(self, record: dict) -> Union[dict, list]:
-        records = []
-        for perm in record["permissions"]:
-            if "entries" not in perm:
-                continue
-
-            for entry in perm["entries"]:
-                if "timestamp" in entry:
-                    records.append({
-                        "timestamp": entry["timestamp"],
-                        "module": self.__class__.__name__,
-                        "event": entry["access"],
-                        "data": f"{record['package_name']} access to "
-                                f"{perm['name']}: {entry['access']}",
-                    })
-
-        return records
-
-    def check_indicators(self) -> None:
-        for result in self.results:
-            if self.indicators:
-                ioc = self.indicators.check_app_id(result.get("package_name"))
-                if ioc:
-                    result["matched_indicator"] = ioc
-                    self.detected.append(result)
-                    continue
-
-            for perm in result["permissions"]:
-                if (perm["name"] == "REQUEST_INSTALL_PACKAGES"
-                        and perm["access"] == "allow"):
-                    self.log.info("Package %s with REQUEST_INSTALL_PACKAGES "
-                                  "permission", result["package_name"])
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
     def run(self) -> None:
         self._adb_connect()
         output = self._adb_command("dumpsys appops")
         self._adb_disconnect()
 
-        self.results = parse_dumpsys_appops(output)
+        self.parse(output)
 
-        self.log.info("Extracted a total of %d records from app-ops manager",
-                      len(self.results))
+        self.log.info(
+            "Extracted a total of %d records from app-ops manager", len(self.results)
+        )

mvt/android/modules/adb/dumpsys_battery_daily.py

@@ -4,14 +4,14 @@
 #   https://license.mvt.re/1.1/
 
 import logging
-from typing import Optional, Union
+from typing import Optional
 
-from mvt.android.parsers import parse_dumpsys_battery_daily
+from mvt.android.artifacts.dumpsys_battery_daily import DumpsysBatteryDailyArtifact
 
 from .base import AndroidExtraction
 
 
-class DumpsysBatteryDaily(AndroidExtraction):
+class DumpsysBatteryDaily(DumpsysBatteryDailyArtifact, AndroidExtraction):
     """This module extracts records from battery daily updates."""
 
     def __init__(
@@ -19,40 +19,26 @@ class DumpsysBatteryDaily(AndroidExtraction):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
-
-    def serialize(self, record: dict) -> Union[dict, list]:
-        return {
-            "timestamp": record["from"],
-            "module": self.__class__.__name__,
-            "event": "battery_daily",
-            "data": f"Recorded update of package {record['package_name']} "
-                    f"with vers {record['vers']}"
-        }
-
-    def check_indicators(self) -> None:
-        if not self.indicators:
-            return
-
-        for result in self.results:
-            ioc = self.indicators.check_app_id(result["package_name"])
-            if ioc:
-                result["matched_indicator"] = ioc
-                self.detected.append(result)
-                continue
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
     def run(self) -> None:
         self._adb_connect()
         output = self._adb_command("dumpsys batterystats --daily")
         self._adb_disconnect()
 
-        self.results = parse_dumpsys_battery_daily(output)
+        self.parse(output)
 
-        self.log.info("Extracted %d records from battery daily stats",
-                      len(self.results))
+        self.log.info(
+            "Extracted %d records from battery daily stats", len(self.results)
+        )

mvt/android/modules/adb/dumpsys_battery_history.py

@@ -6,12 +6,12 @@
 import logging
 from typing import Optional
 
-from mvt.android.parsers import parse_dumpsys_battery_history
+from mvt.android.artifacts.dumpsys_battery_history import DumpsysBatteryHistoryArtifact
 
 from .base import AndroidExtraction
 
 
-class DumpsysBatteryHistory(AndroidExtraction):
+class DumpsysBatteryHistory(DumpsysBatteryHistoryArtifact, AndroidExtraction):
     """This module extracts records from battery history events."""
 
     def __init__(
@@ -19,31 +19,24 @@ class DumpsysBatteryHistory(AndroidExtraction):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
-
-    def check_indicators(self) -> None:
-        if not self.indicators:
-            return
-
-        for result in self.results:
-            ioc = self.indicators.check_app_id(result["package_name"])
-            if ioc:
-                result["matched_indicator"] = ioc
-                self.detected.append(result)
-                continue
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
     def run(self) -> None:
         self._adb_connect()
         output = self._adb_command("dumpsys batterystats --history")
         self._adb_disconnect()
 
-        self.results = parse_dumpsys_battery_history(output)
+        self.parse(output)
 
-        self.log.info("Extracted %d records from battery history",
-                      len(self.results))
+        self.log.info("Extracted %d records from battery history", len(self.results))

mvt/android/modules/adb/dumpsys_dbinfo.py

@@ -6,12 +6,12 @@
 import logging
 from typing import Optional
 
-from mvt.android.parsers import parse_dumpsys_dbinfo
+from mvt.android.artifacts.dumpsys_dbinfo import DumpsysDBInfoArtifact
 
 from .base import AndroidExtraction
 
 
-class DumpsysDBInfo(AndroidExtraction):
+class DumpsysDBInfo(DumpsysDBInfoArtifact, AndroidExtraction):
     """This module extracts records from battery daily updates."""
 
     slug = "dumpsys_dbinfo"
@@ -21,33 +21,27 @@ class DumpsysDBInfo(AndroidExtraction):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
-
-    def check_indicators(self) -> None:
-        if not self.indicators:
-            return
-
-        for result in self.results:
-            path = result.get("path", "")
-            for part in path.split("/"):
-                ioc = self.indicators.check_app_id(part)
-                if ioc:
-                    result["matched_indicator"] = ioc
-                    self.detected.append(result)
-                    continue
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
     def run(self) -> None:
         self._adb_connect()
         output = self._adb_command("dumpsys dbinfo")
         self._adb_disconnect()
 
-        self.results = parse_dumpsys_dbinfo(output)
+        self.parse(output)
 
-        self.log.info("Extracted a total of %d records from database information",
-                      len(self.results))
+        self.log.info(
+            "Extracted a total of %d records from database information",
+            len(self.results),
+        )

mvt/android/modules/adb/dumpsys_full.py

@@ -18,13 +18,18 @@ class DumpsysFull(AndroidExtraction):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
     def run(self) -> None:
         self._adb_connect()

mvt/android/modules/adb/dumpsys_receivers.py

@@ -6,18 +6,12 @@
 import logging
 from typing import Optional
 
-from mvt.android.parsers import parse_dumpsys_receiver_resolver_table
+from mvt.android.artifacts.dumpsys_receivers import DumpsysReceiversArtifact
 
 from .base import AndroidExtraction
 
-INTENT_NEW_OUTGOING_SMS = "android.provider.Telephony.NEW_OUTGOING_SMS"
-INTENT_SMS_RECEIVED = "android.provider.Telephony.SMS_RECEIVED"
-INTENT_DATA_SMS_RECEIVED = "android.intent.action.DATA_SMS_RECEIVED"
-INTENT_PHONE_STATE = "android.intent.action.PHONE_STATE"
-INTENT_NEW_OUTGOING_CALL = "android.intent.action.NEW_OUTGOING_CALL"
 
-
-class DumpsysReceivers(AndroidExtraction):
+class DumpsysReceivers(DumpsysReceiversArtifact, AndroidExtraction):
     """This module extracts details on receivers for risky activities."""
 
     def __init__(
@@ -25,49 +19,26 @@ class DumpsysReceivers(AndroidExtraction):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
         self.results = results if results else {}
 
-    def check_indicators(self) -> None:
-        if not self.indicators:
-            return
-
-        for intent, receivers in self.results.items():
-            for receiver in receivers:
-                if intent == INTENT_NEW_OUTGOING_SMS:
-                    self.log.info("Found a receiver to intercept outgoing SMS messages: \"%s\"",
-                                  receiver["receiver"])
-                elif intent == INTENT_SMS_RECEIVED:
-                    self.log.info("Found a receiver to intercept incoming SMS messages: \"%s\"",
-                                  receiver["receiver"])
-                elif intent == INTENT_DATA_SMS_RECEIVED:
-                    self.log.info("Found a receiver to intercept incoming data SMS message: \"%s\"",
-                                  receiver["receiver"])
-                elif intent == INTENT_PHONE_STATE:
-                    self.log.info("Found a receiver monitoring "
-                                  "telephony state/incoming calls: \"%s\"",
-                                  receiver["receiver"])
-                elif intent == INTENT_NEW_OUTGOING_CALL:
-                    self.log.info("Found a receiver monitoring outgoing calls: \"%s\"",
-                                  receiver["receiver"])
-
-                ioc = self.indicators.check_app_id(receiver["package_name"])
-                if ioc:
-                    receiver["matched_indicator"] = ioc
-                    self.detected.append({intent: receiver})
-                    continue
-
     def run(self) -> None:
         self._adb_connect()
 
         output = self._adb_command("dumpsys package")
-        self.results = parse_dumpsys_receiver_resolver_table(output)
+        self.parse(output)
 
         self._adb_disconnect()
+        self.log.info("Extracted receivers for %d intents", len(self.results))

mvt/android/modules/adb/files.py

@@ -30,13 +30,18 @@ class Files(AndroidExtraction):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
         self.full_find = False
 
     def serialize(self, record: dict) -> Union[dict, list, None]:
@@ -53,12 +58,15 @@ class Files(AndroidExtraction):
     def check_indicators(self) -> None:
         for result in self.results:
             if result.get("is_suid"):
-                self.log.warning("Found an SUID file in a non-standard directory \"%s\".",
-                                 result["path"])
+                self.log.warning(
+                    'Found an SUID file in a non-standard directory "%s".',
+                    result["path"],
+                )
 
             if self.indicators and self.indicators.check_file_path(result["path"]):
-                self.log.warning("Found a known suspicous file at path: \"%s\"",
-                                 result["path"])
+                self.log.warning(
+                    'Found a known suspicous file at path: "%s"', result["path"]
+                )
                 self.detected.append(result)
 
     def backup_file(self, file_path: str) -> None:
@@ -73,13 +81,13 @@ class Files(AndroidExtraction):
         local_file_path = os.path.join(local_files_folder, local_file_name)
 
         try:
-            self._adb_download(remote_path=file_path,
-                               local_path=local_file_path)
+            self._adb_download(remote_path=file_path, local_path=local_file_path)
         except Exception:
             pass
         else:
-            self.log.info("Downloaded file %s to local copy at %s",
-                          file_path, local_file_path)
+            self.log.info(
+                "Downloaded file %s to local copy at %s", file_path, local_file_path
+            )
 
     def find_files(self, folder: str) -> None:
         assert isinstance(self.results, list)
@@ -92,20 +100,21 @@ class Files(AndroidExtraction):
                 if len(file_line) < 6:
                     self.log.info("Skipping invalid file info - %s", file_line.rstrip())
                     continue
-                [unix_timestamp, mode, size,
-                 owner, group, full_path] = file_info
+                [unix_timestamp, mode, size, owner, group, full_path] = file_info
                 mod_time = convert_unix_to_iso(unix_timestamp)
 
-                self.results.append({
-                    "path": full_path,
-                    "modified_time": mod_time,
-                    "mode": mode,
-                    "is_suid": (int(mode, 8) & stat.S_ISUID) == 2048,
-                    "is_sgid": (int(mode, 8) & stat.S_ISGID) == 1024,
-                    "size": size,
-                    "owner": owner,
-                    "group": group,
-                })
+                self.results.append(
+                    {
+                        "path": full_path,
+                        "modified_time": mod_time,
+                        "mode": mode,
+                        "is_suid": (int(mode, 8) & stat.S_ISUID) == 2048,
+                        "is_sgid": (int(mode, 8) & stat.S_ISGID) == 1024,
+                        "size": size,
+                        "owner": owner,
+                        "group": group,
+                    }
+                )
         else:
             output = self._adb_command(f"find '{folder}' -type f 2> /dev/null")
             for file_line in output.splitlines():
@@ -123,18 +132,20 @@ class Files(AndroidExtraction):
             self.find_files(tmp_folder)
 
         for entry in self.results:
-            self.log.info("Found file in tmp folder at path %s",
-                          entry.get("path"))
+            self.log.info("Found file in tmp folder at path %s", entry.get("path"))
             self.backup_file(entry.get("path"))
 
         for media_folder in ANDROID_MEDIA_FOLDERS:
             self.find_files(media_folder)
 
-        self.log.info("Found %s files in primary Android tmp and media folders",
-                      len(self.results))
+        self.log.info(
+            "Found %s files in primary Android tmp and media folders", len(self.results)
+        )
 
-        if self.fast_mode:
-            self.log.info("Flag --fast was enabled: skipping full file listing")
+        if self.module_options.get("fast_mode", None):
+            self.log.info(
+                "The `fast_mode` option was enabled: skipping full file listing"
+            )
         else:
             self.log.info("Processing full file listing. This may take a while...")
             self.find_files("/")

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__(
@@ -20,42 +19,25 @@ class Getprop(AndroidExtraction):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
         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.log.info("Extracted %d Android system properties",
-                      len(self.results))
+        self.parse(output)
+        self.log.info("Extracted %d Android system properties", len(self.results))

mvt/android/modules/adb/logcat.py

@@ -18,37 +18,40 @@ class Logcat(AndroidExtraction):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
     def run(self) -> None:
         self._adb_connect()
 
         # Get the current logcat.
-        output = self._adb_command("logcat -d -b all \"*:V\"")
+        output = self._adb_command('logcat -d -b all "*:V"')
         # Get the locat prior to last reboot.
-        last_output = self._adb_command("logcat -L -b all \"*:V\"")
+        last_output = self._adb_command('logcat -L -b all "*:V"')
 
         if self.results_path:
-            logcat_path = os.path.join(self.results_path,
-                                       "logcat.txt")
+            logcat_path = os.path.join(self.results_path, "logcat.txt")
             with open(logcat_path, "w", encoding="utf-8") as handle:
                 handle.write(output)
 
-            self.log.info("Current logcat logs stored at %s",
-                          logcat_path)
+            self.log.info("Current logcat logs stored at %s", logcat_path)
 
-            logcat_last_path = os.path.join(self.results_path,
-                                            "logcat_last.txt")
+            logcat_last_path = os.path.join(self.results_path, "logcat_last.txt")
             with open(logcat_last_path, "w", encoding="utf-8") as handle:
                 handle.write(last_output)
 
-            self.log.info("Logcat logs prior to last reboot stored at %s",
-                          logcat_last_path)
+            self.log.info(
+                "Logcat logs prior to last reboot stored at %s", logcat_last_path
+            )
 
         self._adb_disconnect()

mvt/android/modules/adb/packages.py

@@ -93,59 +93,65 @@ class Packages(AndroidExtraction):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
     def serialize(self, record: dict) -> Union[dict, list]:
         records = []
 
         timestamps = [
-            {
-                "event": "package_install",
-                "timestamp": record["timestamp"]
-            },
+            {"event": "package_install", "timestamp": record["timestamp"]},
             {
                 "event": "package_first_install",
-                "timestamp": record["first_install_time"]
-            },
-            {
-                "event": "package_last_update",
-                "timestamp": record["last_update_time"]
+                "timestamp": record["first_install_time"],
             },
+            {"event": "package_last_update", "timestamp": record["last_update_time"]},
         ]
 
         for timestamp in timestamps:
-            records.append({
-                "timestamp": timestamp["timestamp"],
-                "module": self.__class__.__name__,
-                "event": timestamp["event"],
-                "data": f"{record['package_name']} (system: {record['system']},"
-                        f" third party: {record['third_party']})",
-            })
+            records.append(
+                {
+                    "timestamp": timestamp["timestamp"],
+                    "module": self.__class__.__name__,
+                    "event": timestamp["event"],
+                    "data": f"{record['package_name']} (system: {record['system']},"
+                    f" third party: {record['third_party']})",
+                }
+            )
 
         return records
 
     def check_indicators(self) -> None:
         for result in self.results:
             if result["package_name"] in ROOT_PACKAGES:
-                self.log.warning("Found an installed package related to "
-                                 "rooting/jailbreaking: \"%s\"",
-                                 result["package_name"])
+                self.log.warning(
+                    "Found an installed package related to "
+                    'rooting/jailbreaking: "%s"',
+                    result["package_name"],
+                )
                 self.detected.append(result)
                 continue
 
             if result["package_name"] in SECURITY_PACKAGES and result["disabled"]:
-                self.log.warning("Found a security package disabled: \"%s\"",
-                                 result["package_name"])
+                self.log.warning(
+                    'Found a security package disabled: "%s"', result["package_name"]
+                )
 
             if result["package_name"] in SYSTEM_UPDATE_PACKAGES and result["disabled"]:
-                self.log.warning("System OTA update package \"%s\" disabled on the phone",
-                                 result["package_name"])
+                self.log.warning(
+                    'System OTA update package "%s" disabled on the phone',
+                    result["package_name"],
+                )
 
             if not self.indicators:
                 continue
@@ -239,22 +245,24 @@ class Packages(AndroidExtraction):
         for file_path in output.splitlines():
             file_path = file_path.strip()
 
-            md5 = self._adb_command(
-                f"md5sum {file_path}").split(" ", maxsplit=1)[0]
-            sha1 = self._adb_command(
-                f"sha1sum {file_path}").split(" ", maxsplit=1)[0]
-            sha256 = self._adb_command(
-                f"sha256sum {file_path}").split(" ", maxsplit=1)[0]
-            sha512 = self._adb_command(
-                f"sha512sum {file_path}").split(" ", maxsplit=1)[0]
+            md5 = self._adb_command(f"md5sum {file_path}").split(" ", maxsplit=1)[0]
+            sha1 = self._adb_command(f"sha1sum {file_path}").split(" ", maxsplit=1)[0]
+            sha256 = self._adb_command(f"sha256sum {file_path}").split(" ", maxsplit=1)[
+                0
+            ]
+            sha512 = self._adb_command(f"sha512sum {file_path}").split(" ", maxsplit=1)[
+                0
+            ]
 
-            package_files.append({
-                "path": file_path,
-                "md5": md5,
-                "sha1": sha1,
-                "sha256": sha256,
-                "sha512": sha512,
-            })
+            package_files.append(
+                {
+                    "path": file_path,
+                    "md5": md5,
+                    "sha1": sha1,
+                    "sha256": sha256,
+                    "sha512": sha512,
+                }
+            )
 
         return package_files
 
@@ -290,8 +298,7 @@ class Packages(AndroidExtraction):
                 "files": package_files,
             }
 
-            dumpsys_package = self._adb_command(
-                f"dumpsys package {package_name}")
+            dumpsys_package = self._adb_command(f"dumpsys package {package_name}")
             package_details = self.parse_package_for_details(dumpsys_package)
             new_package.update(package_details)
 
@@ -324,10 +331,12 @@ class Packages(AndroidExtraction):
                     dangerous_permissions_count += 1
 
             if dangerous_permissions_count >= DANGEROUS_PERMISSIONS_THRESHOLD:
-                self.log.info("Third-party package \"%s\" requested %d "
-                              "potentially dangerous permissions",
-                              result["package_name"],
-                              dangerous_permissions_count)
+                self.log.info(
+                    'Third-party package "%s" requested %d '
+                    "potentially dangerous permissions",
+                    result["package_name"],
+                    dangerous_permissions_count,
+                )
 
         packages_to_lookup = []
         for result in self.results:
@@ -335,14 +344,18 @@ class Packages(AndroidExtraction):
                 continue
 
             packages_to_lookup.append(result)
-            self.log.info("Found non-system package with name \"%s\" installed by \"%s\" on %s",
-                          result["package_name"], result["installer"],
-                          result["timestamp"])
+            self.log.info(
+                'Found non-system package with name "%s" installed by "%s" on %s',
+                result["package_name"],
+                result["installer"],
+                result["timestamp"],
+            )
 
-        if not self.fast_mode:
+        if not self.module_options.get("fast_mode", None):
             self.check_virustotal(packages_to_lookup)
 
-        self.log.info("Extracted at total of %d installed package names",
-                      len(self.results))
+        self.log.info(
+            "Extracted at total of %d installed package names", len(self.results)
+        )
 
         self._adb_disconnect()

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__(
@@ -17,70 +19,24 @@ class Processes(AndroidExtraction):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, 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)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
     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))
+        self.log.info("Extracted records on a total of %d processes", len(self.results))

mvt/android/modules/adb/root_binaries.py

@@ -17,13 +17,23 @@ class RootBinaries(AndroidExtraction):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
+
+    def check_indicators(self) -> None:
+        for root_binary in self.results:
+            self.detected.append(root_binary)
+            self.log.warning('Found root binary "%s"', root_binary)
 
     def run(self) -> None:
         root_binaries = [
@@ -55,7 +65,6 @@ class RootBinaries(AndroidExtraction):
             if "which: not found" in output:
                 continue
 
-            self.detected.append(root_binary)
-            self.log.warning("Found root binary \"%s\"", root_binary)
+            self.results.append(root_binary)
 
         self._adb_disconnect()

mvt/android/modules/adb/selinux_status.py

@@ -19,13 +19,18 @@ class SELinuxStatus(AndroidExtraction):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
         self.results = {} if not results else results
 
@@ -40,4 +45,4 @@ class SELinuxStatus(AndroidExtraction):
         if status == "enforcing":
             self.log.info("SELinux is being regularly enforced")
         else:
-            self.log.warning("SELinux status is \"%s\"!", status)
+            self.log.warning('SELinux status is "%s"!', status)

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__(
@@ -65,27 +19,21 @@ class Settings(AndroidExtraction):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
         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/adb/sms.py

@@ -8,8 +8,7 @@ import os
 import sqlite3
 from typing import Optional, Union
 
-from mvt.android.parsers.backup import (AndroidBackupParsingError,
-                                        parse_tar_for_sms)
+from mvt.android.parsers.backup import AndroidBackupParsingError, parse_tar_for_sms
 from mvt.common.module import InsufficientPrivileges
 from mvt.common.utils import check_for_links, convert_unix_to_iso
 
@@ -50,13 +49,18 @@ class SMS(AndroidExtraction):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
         self.sms_db_type = 0
 
@@ -66,7 +70,7 @@ class SMS(AndroidExtraction):
             "timestamp": record["isodate"],
             "module": self.__class__.__name__,
             "event": f"sms_{record['direction']}",
-            "data": f"{record.get('address', 'unknown source')}: \"{body}\""
+            "data": f"{record.get('address', 'unknown source')}: \"{body}\"",
         }
 
     def check_indicators(self) -> None:
@@ -105,7 +109,7 @@ class SMS(AndroidExtraction):
             for index, value in enumerate(item):
                 message[names[index]] = value
 
-            message["direction"] = ("received" if message["incoming"] == 1 else "sent")
+            message["direction"] = "received" if message["incoming"] == 1 else "sent"
             message["isodate"] = convert_unix_to_iso(message["timestamp"])
 
             # Extract links in the message body
@@ -117,8 +121,7 @@ class SMS(AndroidExtraction):
         cur.close()
         conn.close()
 
-        self.log.info("Extracted a total of %d SMS messages",
-                      len(self.results))
+        self.log.info("Extracted a total of %d SMS messages", len(self.results))
 
     def _extract_sms_adb(self) -> None:
         """Use the Android backup command to extract SMS data from the native
@@ -135,13 +138,14 @@ class SMS(AndroidExtraction):
         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")
+            self.log.info(
+                "Impossible to read SMS from the Android Backup, "
+                "please extract the SMS and try extracting it with "
+                "Android Backup Extractor"
+            )
             return
 
-        self.log.info("Extracted a total of %d SMS messages",
-                      len(self.results))
+        self.log.info("Extracted a total of %d SMS messages", len(self.results))
 
     def run(self) -> None:
         self._adb_connect()
@@ -149,20 +153,24 @@ class SMS(AndroidExtraction):
         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)
+                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)
+                self._adb_process_file(
+                    os.path.join("/", SMS_MMSSMS_PATH), self._parse_db
+                )
 
             self._adb_disconnect()
             return
         except InsufficientPrivileges:
             pass
 
-        self.log.info("No SMS database found. Trying extraction of SMS data "
-                      "using Android backup feature.")
+        self.log.info(
+            "No SMS database found. Trying extraction of SMS data "
+            "using Android backup feature."
+        )
         self._extract_sms_adb()
 
         self._adb_disconnect()

mvt/android/modules/adb/whatsapp.py

@@ -24,13 +24,18 @@ class Whatsapp(AndroidExtraction):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
     def serialize(self, record: dict) -> Union[dict, list]:
         text = record["data"].replace("\n", "\\n")
@@ -38,7 +43,7 @@ class Whatsapp(AndroidExtraction):
             "timestamp": record["isodate"],
             "module": self.__class__.__name__,
             "event": f"whatsapp_msg_{record['direction']}",
-            "data": f"\"{text}\""
+            "data": f'"{text}"',
         }
 
     def check_indicators(self) -> None:
@@ -61,9 +66,11 @@ class Whatsapp(AndroidExtraction):
         """
         conn = sqlite3.connect(db_path)
         cur = conn.cursor()
-        cur.execute("""
+        cur.execute(
+            """
             SELECT * FROM messages;
-        """)
+        """
+        )
         names = [description[0] for description in cur.description]
 
         messages = []
@@ -75,32 +82,30 @@ class Whatsapp(AndroidExtraction):
             if not message["data"]:
                 continue
 
-            message["direction"] = ("send" if message["key_from_me"] == 1 else "received")
+            message["direction"] = "send" if message["key_from_me"] == 1 else "received"
             message["isodate"] = convert_unix_to_iso(message["timestamp"])
 
             # If we find links in the messages or if they are empty we add them
             # to the list.
-            if (check_for_links(message["data"])
-                    or message["data"].strip() == ""):
+            if check_for_links(message["data"]) or message["data"].strip() == "":
                 if message.get("thumb_image"):
-                    message["thumb_image"] = base64.b64encode(
-                        message["thumb_image"])
+                    message["thumb_image"] = base64.b64encode(message["thumb_image"])
 
                 messages.append(message)
 
         cur.close()
         conn.close()
 
-        self.log.info("Extracted a total of %d WhatsApp messages containing links",
-                      len(messages))
+        self.log.info(
+            "Extracted a total of %d WhatsApp messages containing links", len(messages)
+        )
         self.results = messages
 
     def run(self) -> None:
         self._adb_connect()
 
         try:
-            self._adb_process_file(os.path.join("/", WHATSAPP_PATH),
-                                                self._parse_db)
+            self._adb_process_file(os.path.join("/", WHATSAPP_PATH), self._parse_db)
         except Exception as exc:
             self.log.error(exc)
 

mvt/android/modules/androidqf/__init__.py

@@ -6,6 +6,9 @@
 from .dumpsys_accessibility import DumpsysAccessibility
 from .dumpsys_activities import DumpsysActivities
 from .dumpsys_appops import DumpsysAppops
+from .dumpsys_battery_daily import DumpsysBatteryDaily
+from .dumpsys_battery_history import DumpsysBatteryHistory
+from .dumpsys_dbinfo import DumpsysDBInfo
 from .dumpsys_packages import DumpsysPackages
 from .dumpsys_receivers import DumpsysReceivers
 from .getprop import Getprop
@@ -13,6 +16,17 @@ from .processes import Processes
 from .settings import Settings
 from .sms import SMS
 
-ANDROIDQF_MODULES = [DumpsysActivities, DumpsysReceivers, DumpsysAccessibility,
-                     DumpsysAppops, Processes, Getprop, Settings, SMS,
-                     DumpsysPackages]
+ANDROIDQF_MODULES = [
+    DumpsysActivities,
+    DumpsysReceivers,
+    DumpsysAccessibility,
+    DumpsysAppops,
+    DumpsysDBInfo,
+    DumpsysBatteryDaily,
+    DumpsysBatteryHistory,
+    Processes,
+    Getprop,
+    Settings,
+    SMS,
+    DumpsysPackages,
+]

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
@@ -19,20 +20,40 @@ class AndroidQFModule(MVTModule):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Union[List[Dict[str, Any]], Dict[str, Any], None] = None
+        results: Union[List[Dict[str, Any]], Dict[str, Any], None] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            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

@@ -6,12 +6,12 @@
 import logging
 from typing import Optional
 
-from mvt.android.parsers import parse_dumpsys_accessibility
+from mvt.android.artifacts.dumpsys_accessibility import DumpsysAccessibilityArtifact
 
 from .base import AndroidQFModule
 
 
-class DumpsysAccessibility(AndroidQFModule):
+class DumpsysAccessibility(DumpsysAccessibilityArtifact, AndroidQFModule):
     """This module analyse dumpsys accessbility"""
 
     def __init__(
@@ -19,50 +19,33 @@ class DumpsysAccessibility(AndroidQFModule):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
-
-    def check_indicators(self) -> None:
-        if not self.indicators:
-            return
-
-        for result in self.results:
-            ioc = self.indicators.check_app_id(result["package_name"])
-            if ioc:
-                result["matched_indicator"] = ioc
-                self.detected.append(result)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
     def run(self) -> None:
         dumpsys_file = self._get_files_by_pattern("*/dumpsys.txt")
         if not dumpsys_file:
             return
 
-        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
-
-                if not in_accessibility:
-                    continue
-
-                if line.strip().startswith("-------------------------------------------------------------------------------"):  # pylint: disable=line-too-long
-                    break
-
-                lines.append(line.rstrip())
-
-        self.results = parse_dumpsys_accessibility("\n".join(lines))
+        data = self._get_file_content(dumpsys_file[0]).decode("utf-8", errors="replace")
+        content = self.extract_dumpsys_section(data, "DUMP OF SERVICE accessibility:")
+        self.parse(content)
 
         for result in self.results:
-            self.log.info("Found installed accessibility service \"%s\"",
-                          result.get("service"))
+            self.log.info(
+                'Found installed accessibility service "%s"', result.get("service")
+            )
 
-        self.log.info("Identified a total of %d accessibility services",
-                      len(self.results))
+        self.log.info(
+            "Identified a total of %d accessibility services", len(self.results)
+        )

mvt/android/modules/androidqf/dumpsys_activities.py

@@ -6,12 +6,14 @@
 import logging
 from typing import Optional
 
-from mvt.android.parsers import parse_dumpsys_activity_resolver_table
+from mvt.android.artifacts.dumpsys_package_activities import (
+    DumpsysPackageActivitiesArtifact,
+)
 
 from .base import AndroidQFModule
 
 
-class DumpsysActivities(AndroidQFModule):
+class DumpsysActivities(DumpsysPackageActivitiesArtifact, AndroidQFModule):
     """This module extracts details on receivers for risky activities."""
 
     def __init__(
@@ -19,48 +21,30 @@ class DumpsysActivities(AndroidQFModule):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
-        self.results = results if results else {}
-
-    def check_indicators(self) -> None:
-        if not self.indicators:
-            return
-
-        for intent, activities in self.results.items():
-            for activity in activities:
-                ioc = self.indicators.check_app_id(activity["package_name"])
-                if ioc:
-                    activity["matched_indicator"] = ioc
-                    self.detected.append({intent: activity})
+        self.results = results if results else []
 
     def run(self) -> None:
         dumpsys_file = self._get_files_by_pattern("*/dumpsys.txt")
         if not dumpsys_file:
             return
 
-        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
+        # Get data and extract the dumpsys section
+        data = self._get_file_content(dumpsys_file[0]).decode("utf-8", errors="replace")
+        content = self.extract_dumpsys_section(data, "DUMP OF SERVICE package:")
+        # Parse it
+        self.parse(content)
 
-                if not in_package:
-                    continue
-
-                if line.strip().startswith("------------------------------------------------------------------------------"):  # pylint: disable=line-too-long
-                    break
-
-                lines.append(line.rstrip())
-
-        self.results = parse_dumpsys_activity_resolver_table("\n".join(lines))
-
-        self.log.info("Extracted activities for %d intents", len(self.results))
+        self.log.info("Extracted %d package activities", len(self.results))

mvt/android/modules/androidqf/dumpsys_appops.py

@@ -4,80 +4,43 @@
 #   https://license.mvt.re/1.1/
 
 import logging
-from typing import Optional, Union
+from typing import Optional
 
-from mvt.android.parsers import parse_dumpsys_appops
+from mvt.android.artifacts.dumpsys_appops import DumpsysAppopsArtifact
 
 from .base import AndroidQFModule
 
 
-class DumpsysAppops(AndroidQFModule):
-
+class DumpsysAppops(DumpsysAppopsArtifact, AndroidQFModule):
     def __init__(
         self,
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
-
-    def serialize(self, record: dict) -> Union[dict, list]:
-        records = []
-        for perm in record["permissions"]:
-            if "entries" not in perm:
-                continue
-
-            for entry in perm["entries"]:
-                if "timestamp" in entry:
-                    records.append({
-                        "timestamp": entry["timestamp"],
-                        "module": self.__class__.__name__,
-                        "event": entry["access"],
-                        "data": f"{record['package_name']} access to "
-                                f"{perm['name']} : {entry['access']}",
-                    })
-
-        return records
-
-    def check_indicators(self) -> None:
-        for result in self.results:
-            if self.indicators:
-                ioc = self.indicators.check_app_id(result.get("package_name"))
-                if ioc:
-                    result["matched_indicator"] = ioc
-                    self.detected.append(result)
-                    continue
-
-            for perm in result["permissions"]:
-                if (perm["name"] == "REQUEST_INSTALL_PACKAGES"
-                        and perm["access"] == "allow"):
-                    self.log.info("Package %s with REQUEST_INSTALL_PACKAGES permission",
-                                  result["package_name"])
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
     def run(self) -> None:
         dumpsys_file = self._get_files_by_pattern("*/dumpsys.txt")
         if not dumpsys_file:
             return
 
-        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
+        # Extract section
+        data = self._get_file_content(dumpsys_file[0])
+        section = self.extract_dumpsys_section(
+            data.decode("utf-8", errors="replace"), "DUMP OF SERVICE appops:"
+        )
 
-                if in_package:
-                    if line.startswith("-------------------------------------------------------------------------------"):  # pylint: disable=line-too-long
-                        break
-
-                    lines.append(line.rstrip())
-
-        self.results = parse_dumpsys_appops("\n".join(lines))
-        self.log.info("Identified %d applications in AppOps Manager",
-                      len(self.results))
+        # Parse it
+        self.parse(section)
+        self.log.info("Identified %d applications in AppOps Manager", len(self.results))

mvt/android/modules/androidqf/dumpsys_battery_daily.py

@@ -0,0 +1,46 @@
+# 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 logging
+from typing import Optional
+
+from mvt.android.artifacts.dumpsys_battery_daily import DumpsysBatteryDailyArtifact
+
+from .base import AndroidQFModule
+
+
+class DumpsysBatteryDaily(DumpsysBatteryDailyArtifact, AndroidQFModule):
+    def __init__(
+        self,
+        file_path: Optional[str] = None,
+        target_path: Optional[str] = None,
+        results_path: Optional[str] = None,
+        module_options: Optional[dict] = None,
+        log: logging.Logger = logging.getLogger(__name__),
+        results: Optional[list] = None,
+    ) -> None:
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
+
+    def run(self) -> None:
+        dumpsys_file = self._get_files_by_pattern("*/dumpsys.txt")
+        if not dumpsys_file:
+            return
+
+        # Extract section
+        data = self._get_file_content(dumpsys_file[0])
+        section = self.extract_dumpsys_section(
+            data.decode("utf-8", errors="replace"), "DUMP OF SERVICE batterystats:"
+        )
+
+        # Parse it
+        self.parse(section)
+        self.log.info("Extracted a total of %d battery daily stats", len(self.results))

mvt/android/modules/androidqf/dumpsys_battery_history.py

@@ -0,0 +1,46 @@
+# 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 logging
+from typing import Optional
+
+from mvt.android.artifacts.dumpsys_battery_history import DumpsysBatteryHistoryArtifact
+
+from .base import AndroidQFModule
+
+
+class DumpsysBatteryHistory(DumpsysBatteryHistoryArtifact, AndroidQFModule):
+    def __init__(
+        self,
+        file_path: Optional[str] = None,
+        target_path: Optional[str] = None,
+        results_path: Optional[str] = None,
+        module_options: Optional[dict] = None,
+        log: logging.Logger = logging.getLogger(__name__),
+        results: Optional[list] = None,
+    ) -> None:
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
+
+    def run(self) -> None:
+        dumpsys_file = self._get_files_by_pattern("*/dumpsys.txt")
+        if not dumpsys_file:
+            return
+
+        # Extract section
+        data = self._get_file_content(dumpsys_file[0])
+        section = self.extract_dumpsys_section(
+            data.decode("utf-8", errors="replace"), "DUMP OF SERVICE batterystats:"
+        )
+
+        # Parse it
+        self.parse(section)
+        self.log.info("Extracted a total of %d battery daily stats", len(self.results))

mvt/android/modules/androidqf/dumpsys_dbinfo.py

@@ -0,0 +1,46 @@
+# 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 logging
+from typing import Optional
+
+from mvt.android.artifacts.dumpsys_dbinfo import DumpsysDBInfoArtifact
+
+from .base import AndroidQFModule
+
+
+class DumpsysDBInfo(DumpsysDBInfoArtifact, AndroidQFModule):
+    def __init__(
+        self,
+        file_path: Optional[str] = None,
+        target_path: Optional[str] = None,
+        results_path: Optional[str] = None,
+        module_options: Optional[dict] = None,
+        log: logging.Logger = logging.getLogger(__name__),
+        results: Optional[list] = None,
+    ) -> None:
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
+
+    def run(self) -> None:
+        dumpsys_file = self._get_files_by_pattern("*/dumpsys.txt")
+        if not dumpsys_file:
+            return
+
+        # Extract dumpsys DBInfo section
+        data = self._get_file_content(dumpsys_file[0])
+        section = self.extract_dumpsys_section(
+            data.decode("utf-8", errors="replace"), "DUMP OF SERVICE dbinfo:"
+        )
+
+        # Parse it
+        self.parse(section)
+        self.log.info("Identified %d DB Info entries", len(self.results))

mvt/android/modules/androidqf/dumpsys_packages.py

@@ -6,9 +6,11 @@
 import logging
 from typing import Any, Dict, List, Optional, Union
 
-from mvt.android.modules.adb.packages import (DANGEROUS_PERMISSIONS,
-                                              DANGEROUS_PERMISSIONS_THRESHOLD,
-                                              ROOT_PACKAGES)
+from mvt.android.modules.adb.packages import (
+    DANGEROUS_PERMISSIONS,
+    DANGEROUS_PERMISSIONS_THRESHOLD,
+    ROOT_PACKAGES,
+)
 from mvt.android.parsers.dumpsys import parse_dumpsys_packages
 
 from .base import AndroidQFModule
@@ -22,34 +24,43 @@ class DumpsysPackages(AndroidQFModule):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[List[Dict[str, Any]]] = None
+        results: Optional[List[Dict[str, Any]]] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
     def serialize(self, record: dict) -> Union[dict, list]:
         entries = []
         for entry in ["timestamp", "first_install_time", "last_update_time"]:
             if entry in record:
-                entries.append({
-                    "timestamp": record[entry],
-                    "module": self.__class__.__name__,
-                    "event": entry,
-                    "data": f"Package {record['package_name']} "
-                            f"({record['uid']})",
-                })
+                entries.append(
+                    {
+                        "timestamp": record[entry],
+                        "module": self.__class__.__name__,
+                        "event": entry,
+                        "data": f"Package {record['package_name']} "
+                        f"({record['uid']})",
+                    }
+                )
 
         return entries
 
     def check_indicators(self) -> None:
         for result in self.results:
             if result["package_name"] in ROOT_PACKAGES:
-                self.log.warning("Found an installed package related to "
-                                 "rooting/jailbreaking: \"%s\"",
-                                 result["package_name"])
+                self.log.warning(
+                    "Found an installed package related to "
+                    'rooting/jailbreaking: "%s"',
+                    result["package_name"],
+                )
                 self.detected.append(result)
                 continue
 
@@ -67,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
@@ -99,8 +109,10 @@ class DumpsysPackages(AndroidQFModule):
                     dangerous_permissions_count += 1
 
             if dangerous_permissions_count >= DANGEROUS_PERMISSIONS_THRESHOLD:
-                self.log.info("Found package \"%s\" requested %d potentially dangerous permissions",
-                              result["package_name"],
-                              dangerous_permissions_count)
+                self.log.info(
+                    'Found package "%s" requested %d potentially dangerous permissions',
+                    result["package_name"],
+                    dangerous_permissions_count,
+                )
 
         self.log.info("Extracted details on %d packages", len(self.results))

mvt/android/modules/androidqf/dumpsys_receivers.py

@@ -6,15 +6,12 @@
 import logging
 from typing import Any, Dict, List, Optional, Union
 
-from mvt.android.modules.adb.dumpsys_receivers import (
-    INTENT_DATA_SMS_RECEIVED, INTENT_NEW_OUTGOING_CALL,
-    INTENT_NEW_OUTGOING_SMS, INTENT_PHONE_STATE, INTENT_SMS_RECEIVED)
-from mvt.android.parsers import parse_dumpsys_receiver_resolver_table
+from mvt.android.artifacts.dumpsys_receivers import DumpsysReceiversArtifact
 
 from .base import AndroidQFModule
 
 
-class DumpsysReceivers(AndroidQFModule):
+class DumpsysReceivers(DumpsysReceiversArtifact, AndroidQFModule):
     """This module analyse dumpsys receivers"""
 
     def __init__(
@@ -22,65 +19,31 @@ class DumpsysReceivers(AndroidQFModule):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Union[List[Any], Dict[str, Any], None] = None
+        results: Union[List[Any], Dict[str, Any], None] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
         self.results = results if results else {}
 
-    def check_indicators(self) -> None:
-        if not self.indicators:
-            return
-
-        for intent, receivers in self.results.items():
-            for receiver in receivers:
-                if intent == INTENT_NEW_OUTGOING_SMS:
-                    self.log.info("Found a receiver to intercept outgoing SMS messages: \"%s\"",
-                                  receiver["receiver"])
-                elif intent == INTENT_SMS_RECEIVED:
-                    self.log.info("Found a receiver to intercept incoming SMS messages: \"%s\"",
-                                  receiver["receiver"])
-                elif intent == INTENT_DATA_SMS_RECEIVED:
-                    self.log.info("Found a receiver to intercept incoming data SMS message: \"%s\"",
-                                  receiver["receiver"])
-                elif intent == INTENT_PHONE_STATE:
-                    self.log.info("Found a receiver monitoring "
-                                  "telephony state/incoming calls: \"%s\"",
-                                  receiver["receiver"])
-                elif intent == INTENT_NEW_OUTGOING_CALL:
-                    self.log.info("Found a receiver monitoring outgoing calls: \"%s\"",
-                                  receiver["receiver"])
-
-                ioc = self.indicators.check_app_id(receiver["package_name"])
-                if ioc:
-                    receiver["matched_indicator"] = ioc
-                    self.detected.append({intent: receiver})
-
     def run(self) -> None:
         dumpsys_file = self._get_files_by_pattern("*/dumpsys.txt")
         if not dumpsys_file:
             return
+        data = self._get_file_content(dumpsys_file[0])
 
-        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
+        dumpsys_section = self.extract_dumpsys_section(
+            data.decode("utf-8", errors="replace"), "DUMP OF SERVICE package:"
+        )
 
-                if not in_receivers:
-                    continue
-
-                if line.strip().startswith("------------------------------------------------------------------------------"):  # pylint: disable=line-too-long
-                    break
-
-                lines.append(line.rstrip())
-
-        self.results = parse_dumpsys_receiver_resolver_table("\n".join(lines))
+        self.parse(dumpsys_section)
 
         self.log.info("Extracted receivers for %d intents", len(self.results))

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__(
@@ -34,43 +19,27 @@ class Getprop(AndroidQFModule):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
         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__(
@@ -17,76 +19,23 @@ class Processes(AndroidQFModule):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, 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,
-            })
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
     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__(
@@ -19,40 +19,38 @@ class Settings(AndroidQFModule):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
         self.results = {}
 
     def run(self) -> None:
         for setting_file in self._get_files_by_pattern("*/settings_*.txt"):
-            namespace = setting_file[setting_file.rfind("_")+1:-4]
+            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
 
-                    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
-
-        self.log.info("Identified %d settings",
-                      sum([len(val) for val in self.results.values()]))
+        self.log.info(
+            "Identified %d settings", sum([len(val) for val in self.results.values()])
+        )

mvt/android/modules/androidqf/sms.py

@@ -3,13 +3,17 @@
 # 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.parsers.backup import (AndroidBackupParsingError,
-                                        InvalidBackupPassword, parse_ab_header,
-                                        parse_backup_file, parse_tar_for_sms)
+from mvt.android.modules.backup.helpers import prompt_or_load_android_backup_password
+from mvt.android.parsers.backup import (
+    AndroidBackupParsingError,
+    InvalidBackupPassword,
+    parse_ab_header,
+    parse_backup_file,
+    parse_tar_for_sms,
+)
 
 from .base import AndroidQFModule
 
@@ -22,13 +26,18 @@ class SMS(AndroidQFModule):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
     def check_indicators(self) -> None:
         if not self.indicators:
@@ -49,15 +58,23 @@ 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:
             self.log.critical("Invalid backup password")
             return
         except AndroidBackupParsingError:
-            self.log.critical("Impossible to parse this backup file, please use"
-                              " Android Backup Extractor instead")
+            self.log.critical(
+                "Impossible to parse this backup file, please use"
+                " Android Backup Extractor instead"
+            )
             return
 
         if not tardata:
@@ -66,9 +83,11 @@ class SMS(AndroidQFModule):
         try:
             self.results = parse_tar_for_sms(tardata)
         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")
+            self.log.info(
+                "Impossible to read SMS from the Android Backup, "
+                "please extract the SMS and try extracting it with "
+                "Android Backup Extractor"
+            )
             return
 
     def run(self) -> None:
@@ -77,9 +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.log.info("Identified %d SMS in backup data",
-                      len(self.results))
+        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/base.py

@@ -20,13 +20,18 @@ class BackupExtraction(MVTModule):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
         self.ab = None
         self.backup_path = None
         self.tar = None
@@ -39,7 +44,9 @@ class BackupExtraction(MVTModule):
         self.backup_path = backup_path
         self.files = files
 
-    def from_ab(self, file_path: Optional[str], tar: Optional[TarFile], files: List[str]) -> None:
+    def from_ab(
+        self, file_path: Optional[str], tar: Optional[TarFile], files: List[str]
+    ) -> None:
         """
         Extract the files
         """

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/backup/sms.py

@@ -12,19 +12,23 @@ from mvt.common.utils import check_for_links
 
 
 class SMS(BackupExtraction):
-
     def __init__(
         self,
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
         self.results = []
 
     def check_indicators(self) -> None:
@@ -55,5 +59,4 @@ class SMS(BackupExtraction):
             data = self._get_file_content(file)
             self.results.extend(parse_sms_file(data))
 
-        self.log.info("Extracted a total of %d SMS & MMS messages",
-                      len(self.results))
+        self.log.info("Extracted a total of %d SMS & MMS messages", len(self.results))

mvt/android/modules/bugreport/__init__.py

@@ -13,5 +13,14 @@ from .getprop import Getprop
 from .packages import Packages
 from .receivers import Receivers
 
-BUGREPORT_MODULES = [Accessibility, Activities, Appops, BatteryDaily,
-                     BatteryHistory, DBInfo, Getprop, Packages, Receivers]
+BUGREPORT_MODULES = [
+    Accessibility,
+    Activities,
+    Appops,
+    BatteryDaily,
+    BatteryHistory,
+    DBInfo,
+    Getprop,
+    Packages,
+    Receivers,
+]

mvt/android/modules/bugreport/accessibility.py

@@ -6,12 +6,12 @@
 import logging
 from typing import Optional
 
-from mvt.android.parsers import parse_dumpsys_accessibility
+from mvt.android.artifacts.dumpsys_accessibility import DumpsysAccessibilityArtifact
 
 from .base import BugReportModule
 
 
-class Accessibility(BugReportModule):
+class Accessibility(DumpsysAccessibilityArtifact, BugReportModule):
     """This module extracts stats on accessibility."""
 
     def __init__(
@@ -19,51 +19,39 @@ class Accessibility(BugReportModule):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
-
-    def check_indicators(self) -> None:
-        if not self.indicators:
-            return
-
-        for result in self.results:
-            ioc = self.indicators.check_app_id(result["package_name"])
-            if ioc:
-                result["matched_indicator"] = ioc
-                self.detected.append(result)
-                continue
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
     def run(self) -> None:
-        content = self._get_dumpstate_file()
-        if not content:
-            self.log.error("Unable to find dumpstate file. "
-                           "Did you provide a valid bug report archive?")
+        full_dumpsys = self._get_dumpstate_file()
+        if not full_dumpsys:
+            self.log.error(
+                "Unable to find dumpstate file. "
+                "Did you provide a valid bug report archive?"
+            )
             return
 
-        lines = []
-        in_accessibility = False
-        for line in content.decode(errors="ignore").splitlines():
-            if line.strip() == "DUMP OF SERVICE accessibility:":
-                in_accessibility = True
-                continue
+        content = self.extract_dumpsys_section(
+            full_dumpsys.decode("utf-8", errors="ignore"),
+            "DUMP OF SERVICE accessibility:",
+        )
+        self.parse(content)
 
-            if not in_accessibility:
-                continue
-
-            if line.strip().startswith("------------------------------------------------------------------------------"):  # pylint: disable=line-too-long
-                break
-
-            lines.append(line)
-
-        self.results = parse_dumpsys_accessibility("\n".join(lines))
         for result in self.results:
-            self.log.info("Found installed accessibility service \"%s\"",
-                          result.get("service"))
+            self.log.info(
+                'Found installed accessibility service "%s"', result.get("service")
+            )
 
-        self.log.info("Identified a total of %d accessibility services",
-                      len(self.results))
+        self.log.info(
+            "Identified a total of %d accessibility services", len(self.results)
+        )

mvt/android/modules/bugreport/activities.py

@@ -6,12 +6,14 @@
 import logging
 from typing import Optional
 
-from mvt.android.parsers import parse_dumpsys_activity_resolver_table
+from mvt.android.artifacts.dumpsys_package_activities import (
+    DumpsysPackageActivitiesArtifact,
+)
 
 from .base import BugReportModule
 
 
-class Activities(BugReportModule):
+class Activities(DumpsysPackageActivitiesArtifact, BugReportModule):
     """This module extracts details on receivers for risky activities."""
 
     def __init__(
@@ -19,50 +21,36 @@ class Activities(BugReportModule):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
-        self.results = results if results else {}
-
-    def check_indicators(self) -> None:
-        if not self.indicators:
-            return
-
-        for intent, activities in self.results.items():
-            for activity in activities:
-                ioc = self.indicators.check_app_id(activity["package_name"])
-                if ioc:
-                    activity["matched_indicator"] = ioc
-                    self.detected.append({intent: activity})
-                    continue
+        self.results = results if results else []
 
     def run(self) -> None:
         content = self._get_dumpstate_file()
         if not content:
-            self.log.error("Unable to find dumpstate file. "
-                           "Did you provide a valid bug report archive?")
+            self.log.error(
+                "Unable to find dumpstate file. "
+                "Did you provide a valid bug report archive?"
+            )
             return
 
-        lines = []
-        in_package = False
-        for line in content.decode(errors="ignore").splitlines():
-            if line.strip() == "DUMP OF SERVICE package:":
-                in_package = True
-                continue
+        # Extract package section
+        section = self.extract_dumpsys_section(
+            content.decode("utf-8", errors="ignore"), "DUMP OF SERVICE package:"
+        )
 
-            if not in_package:
-                continue
+        # Parse
+        self.parse(section)
 
-            if line.strip().startswith("------------------------------------------------------------------------------"):  # pylint: disable=line-too-long
-                break
-
-            lines.append(line)
-
-        self.results = parse_dumpsys_activity_resolver_table("\n".join(lines))
-
-        self.log.info("Extracted activities for %d intents", len(self.results))
+        self.log.info("Extracted %d package activities", len(self.results))

mvt/android/modules/bugreport/appops.py

@@ -4,14 +4,14 @@
 #   https://license.mvt.re/1.1/
 
 import logging
-from typing import Optional, Union
+from typing import Optional
 
-from mvt.android.parsers import parse_dumpsys_appops
+from mvt.android.artifacts.dumpsys_appops import DumpsysAppopsArtifact
 
 from .base import BugReportModule
 
 
-class Appops(BugReportModule):
+class Appops(DumpsysAppopsArtifact, BugReportModule):
     """This module extracts information on package from App-Ops Manager."""
 
     def __init__(
@@ -19,70 +19,33 @@ class Appops(BugReportModule):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
-
-    def serialize(self, record: dict) -> Union[dict, list]:
-        records = []
-        for perm in record["permissions"]:
-            if "entries" not in perm:
-                continue
-
-            for entry in perm["entries"]:
-                if "timestamp" in entry:
-                    records.append({
-                        "timestamp": entry["timestamp"],
-                        "module": self.__class__.__name__,
-                        "event": entry["access"],
-                        "data": f"{record['package_name']} access to "
-                                f"{perm['name']}: {entry['access']}",
-                    })
-
-        return records
-
-    def check_indicators(self) -> None:
-        for result in self.results:
-            if self.indicators:
-                ioc = self.indicators.check_app_id(result.get("package_name"))
-                if ioc:
-                    result["matched_indicator"] = ioc
-                    self.detected.append(result)
-                    continue
-
-            for perm in result["permissions"]:
-                if (perm["name"] == "REQUEST_INSTALL_PACKAGES"
-                        and perm["access"] == "allow"):
-                    self.log.info("Package %s with REQUEST_INSTALL_PACKAGES permission",
-                                  result["package_name"])
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
     def run(self) -> None:
         content = self._get_dumpstate_file()
         if not content:
-            self.log.error("Unable to find dumpstate file. "
-                           "Did you provide a valid bug report archive?")
+            self.log.error(
+                "Unable to find dumpstate file. "
+                "Did you provide a valid bug report archive?"
+            )
             return
 
-        lines = []
-        in_appops = False
-        for line in content.decode(errors="ignore").splitlines():
-            if line.strip() == "DUMP OF SERVICE appops:":
-                in_appops = True
-                continue
+        section = self.extract_dumpsys_section(
+            content.decode("utf-8", errors="replace"), "DUMP OF SERVICE appops:"
+        )
+        self.parse(section)
 
-            if not in_appops:
-                continue
-
-            if line.strip().startswith("------------------------------------------------------------------------------"):  # pylint: disable=line-too-long
-                break
-
-            lines.append(line)
-
-        self.results = parse_dumpsys_appops("\n".join(lines))
-
-        self.log.info("Identified a total of %d packages in App-Ops Manager",
-                      len(self.results))
+        self.log.info(
+            "Identified a total of %d packages in App-Ops Manager", len(self.results)
+        )

mvt/android/modules/bugreport/base.py

@@ -20,20 +20,27 @@ class BugReportModule(MVTModule):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
         self.zip_archive: Optional[ZipFile] = None
         self.extract_path: Optional[str] = None
         self.extract_files: List[str] = []
         self.zip_files: List[str] = []
 
-    def from_folder(self, extract_path: Optional[str], extract_files: List[str]) -> None:
+    def from_folder(
+        self, extract_path: Optional[str], extract_files: List[str]
+    ) -> None:
         self.extract_path = extract_path
         self.extract_files = extract_files
 

mvt/android/modules/bugreport/battery_daily.py

@@ -4,14 +4,14 @@
 #   https://license.mvt.re/1.1/
 
 import logging
-from typing import Optional, Union
+from typing import Optional
 
-from mvt.android.parsers import parse_dumpsys_battery_daily
+from mvt.android.artifacts.dumpsys_battery_daily import DumpsysBatteryDailyArtifact
 
 from .base import BugReportModule
 
 
-class BatteryDaily(BugReportModule):
+class BatteryDaily(DumpsysBatteryDailyArtifact, BugReportModule):
     """This module extracts records from battery daily updates."""
 
     def __init__(
@@ -19,66 +19,31 @@ class BatteryDaily(BugReportModule):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
-
-    def serialize(self, record: dict) -> Union[dict, list]:
-        return {
-            "timestamp": record["from"],
-            "module": self.__class__.__name__,
-            "event": "battery_daily",
-            "data": f"Recorded update of package {record['package_name']} "
-                    f"with vers {record['vers']}"
-        }
-
-    def check_indicators(self) -> None:
-        if not self.indicators:
-            return
-
-        for result in self.results:
-            ioc = self.indicators.check_app_id(result["package_name"])
-            if ioc:
-                result["matched_indicator"] = ioc
-                self.detected.append(result)
-                continue
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
     def run(self) -> None:
         content = self._get_dumpstate_file()
         if not content:
-            self.log.error("Unable to find dumpstate file. "
-                           "Did you provide a valid bug report archive?")
+            self.log.error(
+                "Unable to find dumpstate file. "
+                "Did you provide a valid bug report archive?"
+            )
             return
 
-        lines = []
-        in_batterystats = False
-        in_daily = False
-        for line in content.decode(errors="ignore").splitlines():
-            if line.strip() == "DUMP OF SERVICE batterystats:":
-                in_batterystats = True
-                continue
+        dumpsys_section = self.extract_dumpsys_section(
+            content.decode("utf-8", errors="replace"), "DUMP OF SERVICE batterystats:"
+        )
+        self.parse(dumpsys_section)
 
-            if not in_batterystats:
-                continue
-
-            if line.strip() == "Daily stats:":
-                lines.append(line)
-                in_daily = True
-                continue
-
-            if not in_daily:
-                continue
-
-            if line.strip() == "":
-                break
-
-            lines.append(line)
-
-        self.results = parse_dumpsys_battery_daily("\n".join(lines))
-
-        self.log.info("Extracted a total of %d battery daily stats",
-                      len(self.results))
+        self.log.info("Extracted a total of %d battery daily stats", len(self.results))

mvt/android/modules/bugreport/battery_history.py

@@ -6,12 +6,12 @@
 import logging
 from typing import Optional
 
-from mvt.android.parsers import parse_dumpsys_battery_history
+from mvt.android.artifacts.dumpsys_battery_history import DumpsysBatteryHistoryArtifact
 
 from .base import BugReportModule
 
 
-class BatteryHistory(BugReportModule):
+class BatteryHistory(DumpsysBatteryHistoryArtifact, BugReportModule):
     """This module extracts records from battery daily updates."""
 
     def __init__(
@@ -19,49 +19,33 @@ class BatteryHistory(BugReportModule):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
-
-    def check_indicators(self) -> None:
-        if not self.indicators:
-            return
-
-        for result in self.results:
-            ioc = self.indicators.check_app_id(result["package_name"])
-            if ioc:
-                result["matched_indicator"] = ioc
-                self.detected.append(result)
-                continue
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
     def run(self) -> None:
         content = self._get_dumpstate_file()
         if not content:
-            self.log.error("Unable to find dumpstate file. "
-                           "Did you provide a valid bug report archive?")
+            self.log.error(
+                "Unable to find dumpstate file. "
+                "Did you provide a valid bug report archive?"
+            )
             return
 
-        lines = []
-        in_history = False
-        for line in content.decode(errors="ignore").splitlines():
-            if line.strip().startswith("Battery History "):
-                lines.append(line)
-                in_history = True
-                continue
+        dumpsys_section = self.extract_dumpsys_section(
+            content.decode("utf-8", errors="replace"), "DUMP OF SERVICE batterystats:"
+        )
+        self.parse(dumpsys_section)
 
-            if not in_history:
-                continue
-
-            if line.strip() == "":
-                break
-
-            lines.append(line)
-
-        self.results = parse_dumpsys_battery_history("\n".join(lines))
-
-        self.log.info("Extracted a total of %d battery history records",
-                      len(self.results))
+        self.log.info(
+            "Extracted a total of %d battery history records", len(self.results)
+        )

mvt/android/modules/bugreport/dbinfo.py

@@ -6,12 +6,12 @@
 import logging
 from typing import Optional
 
-from mvt.android.parsers import parse_dumpsys_dbinfo
+from mvt.android.artifacts.dumpsys_dbinfo import DumpsysDBInfoArtifact
 
 from .base import BugReportModule
 
 
-class DBInfo(BugReportModule):
+class DBInfo(DumpsysDBInfoArtifact, BugReportModule):
     """This module extracts records from battery daily updates."""
 
     slug = "dbinfo"
@@ -21,50 +21,34 @@ class DBInfo(BugReportModule):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
-
-    def check_indicators(self) -> None:
-        if not self.indicators:
-            return
-
-        for result in self.results:
-            path = result.get("path", "")
-            for part in path.split("/"):
-                ioc = self.indicators.check_app_id(part)
-                if ioc:
-                    result["matched_indicator"] = ioc
-                    self.detected.append(result)
-                    continue
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
     def run(self) -> None:
-        content = self._get_dumpstate_file()
-        if not content:
-            self.log.error("Unable to find dumpstate file. "
-                           "Did you provide a valid bug report archive?")
+        data = self._get_dumpstate_file()
+        if not data:
+            self.log.error(
+                "Unable to find dumpstate file. "
+                "Did you provide a valid bug report archive?"
+            )
             return
 
-        in_dbinfo = False
-        lines = []
-        for line in content.decode(errors="ignore").splitlines():
-            if line.strip() == "DUMP OF SERVICE dbinfo:":
-                in_dbinfo = True
-                continue
+        section = self.extract_dumpsys_section(
+            data.decode("utf-8", errors="ignore"), "DUMP OF SERVICE dbinfo:"
+        )
 
-            if not in_dbinfo:
-                continue
-
-            if line.strip().startswith("------------------------------------------------------------------------------"):  # pylint: disable=line-too-long
-                break
-
-            lines.append(line)
-
-        self.results = parse_dumpsys_dbinfo("\n".join(lines))
-
-        self.log.info("Extracted a total of %d database connection pool records",
-                      len(self.results))
+        self.parse(section)
+        self.log.info(
+            "Extracted a total of %d database connection pool records",
+            len(self.results),
+        )

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__(
@@ -20,21 +19,28 @@ class Getprop(BugReportModule):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            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()
         if not content:
-            self.log.error("Unable to find dumpstate file. "
-                           "Did you provide a valid bug report archive?")
+            self.log.error(
+                "Unable to find dumpstate file. "
+                "Did you provide a valid bug report archive?"
+            )
             return
 
         lines = []
@@ -53,17 +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.log.info("Extracted %d Android system properties",
-                      len(self.results))
+        self.parse("\n".join(lines))
+        self.log.info("Extracted %d Android system properties", len(self.results))

mvt/android/modules/bugreport/packages.py

@@ -6,9 +6,11 @@
 import logging
 from typing import Optional, Union
 
-from mvt.android.modules.adb.packages import (DANGEROUS_PERMISSIONS,
-                                              DANGEROUS_PERMISSIONS_THRESHOLD,
-                                              ROOT_PACKAGES)
+from mvt.android.modules.adb.packages import (
+    DANGEROUS_PERMISSIONS,
+    DANGEROUS_PERMISSIONS_THRESHOLD,
+    ROOT_PACKAGES,
+)
 from mvt.android.parsers.dumpsys import parse_dumpsys_packages
 
 from .base import BugReportModule
@@ -22,48 +24,51 @@ class Packages(BugReportModule):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
     def serialize(self, record: dict) -> Union[dict, list]:
         records = []
 
         timestamps = [
-            {
-                "event": "package_install",
-                "timestamp": record["timestamp"]
-            },
+            {"event": "package_install", "timestamp": record["timestamp"]},
             {
                 "event": "package_first_install",
-                "timestamp": record["first_install_time"]
-            },
-            {
-                "event": "package_last_update",
-                "timestamp": record["last_update_time"]
+                "timestamp": record["first_install_time"],
             },
+            {"event": "package_last_update", "timestamp": record["last_update_time"]},
         ]
 
         for timestamp in timestamps:
-            records.append({
-                "timestamp": timestamp["timestamp"],
-                "module": self.__class__.__name__,
-                "event": timestamp["event"],
-                "data": f"Install or update of package {record['package_name']}",
-            })
+            records.append(
+                {
+                    "timestamp": timestamp["timestamp"],
+                    "module": self.__class__.__name__,
+                    "event": timestamp["event"],
+                    "data": f"Install or update of package {record['package_name']}",
+                }
+            )
 
         return records
 
     def check_indicators(self) -> None:
         for result in self.results:
             if result["package_name"] in ROOT_PACKAGES:
-                self.log.warning("Found an installed package related to "
-                                 "rooting/jailbreaking: \"%s\"",
-                                 result["package_name"])
+                self.log.warning(
+                    "Found an installed package related to "
+                    'rooting/jailbreaking: "%s"',
+                    result["package_name"],
+                )
                 self.detected.append(result)
                 continue
 
@@ -79,8 +84,10 @@ class Packages(BugReportModule):
     def run(self) -> None:
         content = self._get_dumpstate_file()
         if not content:
-            self.log.error("Unable to find dumpstate file. "
-                           "Did you provide a valid bug report archive?")
+            self.log.error(
+                "Unable to find dumpstate file. "
+                "Did you provide a valid bug report archive?"
+            )
             return
 
         in_package = False
@@ -115,8 +122,10 @@ class Packages(BugReportModule):
                     dangerous_permissions_count += 1
 
             if dangerous_permissions_count >= DANGEROUS_PERMISSIONS_THRESHOLD:
-                self.log.info("Found package \"%s\" requested %d potentially dangerous permissions",
-                              result["package_name"],
-                              dangerous_permissions_count)
+                self.log.info(
+                    'Found package "%s" requested %d potentially dangerous permissions',
+                    result["package_name"],
+                    dangerous_permissions_count,
+                )
 
         self.log.info("Extracted details on %d packages", len(self.results))

mvt/android/modules/bugreport/receivers.py

@@ -6,18 +6,12 @@
 import logging
 from typing import Optional
 
-from mvt.android.parsers import parse_dumpsys_receiver_resolver_table
+from mvt.android.artifacts.dumpsys_receivers import DumpsysReceiversArtifact
 
 from .base import BugReportModule
 
-INTENT_NEW_OUTGOING_SMS = "android.provider.Telephony.NEW_OUTGOING_SMS"
-INTENT_SMS_RECEIVED = "android.provider.Telephony.SMS_RECEIVED"
-INTENT_DATA_SMS_RECEIVED = "android.intent.action.DATA_SMS_RECEIVED"
-INTENT_PHONE_STATE = "android.intent.action.PHONE_STATE"
-INTENT_NEW_OUTGOING_CALL = "android.intent.action.NEW_OUTGOING_CALL"
 
-
-class Receivers(BugReportModule):
+class Receivers(DumpsysReceiversArtifact, BugReportModule):
     """This module extracts details on receivers for risky activities."""
 
     def __init__(
@@ -25,67 +19,33 @@ class Receivers(BugReportModule):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Optional[list] = None
+        results: Optional[list] = None,
     ) -> None:
-        super().__init__(file_path=file_path, target_path=target_path,
-                         results_path=results_path, fast_mode=fast_mode,
-                         log=log, results=results)
+        super().__init__(
+            file_path=file_path,
+            target_path=target_path,
+            results_path=results_path,
+            module_options=module_options,
+            log=log,
+            results=results,
+        )
 
         self.results = results if results else {}
 
-    def check_indicators(self) -> None:
-        if not self.indicators:
-            return
-
-        for intent, receivers in self.results.items():
-            for receiver in receivers:
-                if intent == INTENT_NEW_OUTGOING_SMS:
-                    self.log.info("Found a receiver to intercept outgoing SMS messages: \"%s\"",
-                                  receiver["receiver"])
-                elif intent == INTENT_SMS_RECEIVED:
-                    self.log.info("Found a receiver to intercept incoming SMS messages: \"%s\"",
-                                  receiver["receiver"])
-                elif intent == INTENT_DATA_SMS_RECEIVED:
-                    self.log.info("Found a receiver to intercept incoming data SMS message: \"%s\"",
-                                  receiver["receiver"])
-                elif intent == INTENT_PHONE_STATE:
-                    self.log.info("Found a receiver monitoring "
-                                  "telephony state/incoming calls: \"%s\"",
-                                  receiver["receiver"])
-                elif intent == INTENT_NEW_OUTGOING_CALL:
-                    self.log.info("Found a receiver monitoring outgoing calls: \"%s\"",
-                                  receiver["receiver"])
-
-                ioc = self.indicators.check_app_id(receiver["package_name"])
-                if ioc:
-                    receiver["matched_indicator"] = ioc
-                    self.detected.append({intent: receiver})
-                    continue
-
     def run(self) -> None:
         content = self._get_dumpstate_file()
         if not content:
-            self.log.error("Unable to find dumpstate file. "
-                           "Did you provide a valid bug report archive?")
+            self.log.error(
+                "Unable to find dumpstate file. "
+                "Did you provide a valid bug report archive?"
+            )
             return
 
-        in_receivers = False
-        lines = []
-        for line in content.decode(errors="ignore").splitlines():
-            if line.strip() == "DUMP OF SERVICE package:":
-                in_receivers = True
-                continue
-
-            if not in_receivers:
-                continue
-
-            if line.strip().startswith("------------------------------------------------------------------------------"):  # pylint: disable=line-too-long
-                break
-
-            lines.append(line)
-
-        self.results = parse_dumpsys_receiver_resolver_table("\n".join(lines))
+        dumpsys_section = self.extract_dumpsys_section(
+            content.decode("utf-8", errors="replace"), "DUMP OF SERVICE package:"
+        )
 
+        self.parse(dumpsys_section)
         self.log.info("Extracted receivers for %d intents", len(self.results))

mvt/android/parsers/__init__.py

@@ -2,10 +2,3 @@
 # 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 .dumpsys import (parse_dumpsys_accessibility,
-                      parse_dumpsys_activity_resolver_table,
-                      parse_dumpsys_appops, parse_dumpsys_battery_daily,
-                      parse_dumpsys_battery_history, parse_dumpsys_dbinfo,
-                      parse_dumpsys_receiver_resolver_table)
-from .getprop import parse_getprop

mvt/android/parsers/backup.py

@@ -31,15 +31,16 @@ class InvalidBackupPassword(AndroidBackupParsingError):
 
 # TODO: Need to clean all the following code and conform it to the coding style.
 
+
 def to_utf8_bytes(input_bytes):
     output = []
     for byte in input_bytes:
-        if byte < ord(b'\x80'):
+        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')))
+            output.append(ord("\xef") | (byte >> 12))
+            output.append(ord("\xbc") | ((byte >> 6) & ord("\x3f")))
+            output.append(ord("\x80") | (byte & ord("\x3f")))
     return bytes(output)
 
 
@@ -55,33 +56,38 @@ def parse_ab_header(data):
             "backup": True,
             "compression": (is_compressed == b"1"),
             "version": int(version),
-            "encryption": encryption.decode("utf-8")
+            "encryption": encryption.decode("utf-8"),
         }
 
-    return {
-        "backup": False,
-        "compression": None,
-        "version": None,
-        "encryption": None
-    }
+    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):
+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)
+    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()
+        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)
@@ -103,8 +109,9 @@ def decrypt_master_key(password, user_salt, user_iv, pbkdf2_rounds,
         hmac_mk = master_key
 
     # Derive checksum to confirm successful backup decryption.
-    kdf = PBKDF2HMAC(algorithm=hashes.SHA1(), length=32, salt=checksum_salt,
-                     iterations=pbkdf2_rounds)
+    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:
@@ -113,8 +120,7 @@ def decrypt_master_key(password, user_salt, user_iv, pbkdf2_rounds,
     return master_key, master_iv
 
 
-def decrypt_backup_data(encrypted_backup, password, encryption_algo,
-                        format_version):
+def decrypt_backup_data(encrypted_backup, password, encryption_algo, format_version):
     """
     Generate encryption keyffrom password and do decryption
 
@@ -125,8 +131,14 @@ def decrypt_backup_data(encrypted_backup, password, encryption_algo,
     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,
+        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"))
@@ -135,13 +147,15 @@ def decrypt_backup_data(encrypted_backup, password, encryption_algo,
     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)
+    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))
@@ -160,21 +174,23 @@ def parse_backup_file(data, password=None):
     if not data.startswith(b"ANDROID BACKUP"):
         raise AndroidBackupParsingError("Invalid file header")
 
-    [_, version, is_compressed,
-     encryption_algo, tar_data] = data.split(b"\n", 4)
+    [_, 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)
+        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 as exc:
-            raise AndroidBackupParsingError("Impossible to decompress the backup file") from exc
+            raise AndroidBackupParsingError(
+                "Impossible to decompress the backup file"
+            ) from exc
 
     return tar_data
 
@@ -189,9 +205,10 @@ def parse_tar_for_sms(data):
     res = []
     with tarfile.open(fileobj=dbytes) as tar:
         for member in tar.getmembers():
-            if (member.name.startswith("apps/com.android.providers.telephony/d_f/")
-                    and (member.name.endswith("_sms_backup")
-                         or member.name.endswith("_mms_backup"))):
+            if member.name.startswith("apps/com.android.providers.telephony/d_f/") and (
+                member.name.endswith("_sms_backup")
+                or member.name.endswith("_mms_backup")
+            ):
                 dhandler = tar.extractfile(member)
                 res.extend(parse_sms_file(dhandler.read()))
 
@@ -216,7 +233,7 @@ def parse_sms_file(data):
         message_links = check_for_links(entry["body"])
 
         entry["isodate"] = convert_unix_to_iso(int(entry["date"]) / 1000)
-        entry["direction"] = ("sent" if int(entry["date_sent"]) else "received")
+        entry["direction"] = "sent" if int(entry["date_sent"]) else "received"
 
         # Extract links from the body
         if message_links or entry["body"].strip() == "":

mvt/android/parsers/dumpsys.py

@@ -4,391 +4,8 @@
 #   https://license.mvt.re/1.1/
 
 import re
-from datetime import datetime
 from typing import Any, Dict, List
 
-from mvt.common.utils import convert_datetime_to_iso
-
-
-def parse_dumpsys_accessibility(output: str) -> List[Dict[str, str]]:
-    results = []
-
-    in_services = False
-    for line in output.splitlines():
-        if line.strip().startswith("installed services:"):
-            in_services = True
-            continue
-
-        if not in_services:
-            continue
-
-        if line.strip() == "}":
-            break
-
-        service = line.split(":")[1].strip()
-
-        results.append({
-            "package_name": service.split("/")[0],
-            "service": service,
-        })
-
-    return results
-
-
-def parse_dumpsys_activity_resolver_table(output: str) -> Dict[str, Any]:
-    results = {}
-
-    in_activity_resolver_table = False
-    in_non_data_actions = False
-    intent = None
-    for line in output.splitlines():
-        if line.startswith("Activity Resolver Table:"):
-            in_activity_resolver_table = True
-            continue
-
-        if not in_activity_resolver_table:
-            continue
-
-        if line.startswith("  Non-Data Actions:"):
-            in_non_data_actions = True
-            continue
-
-        if not in_non_data_actions:
-            continue
-
-        # If we hit an empty line, the Non-Data Actions section should be
-        # finished.
-        if line.strip() == "":
-            break
-
-        # We detect the action name.
-        if (line.startswith(" " * 6) and not line.startswith(" " * 8)
-                and ":" in line):
-            intent = line.strip().replace(":", "")
-            results[intent] = []
-            continue
-
-        # If we are not in an intent block yet, skip.
-        if not intent:
-            continue
-
-        # If we are in a block but the line does not start with 8 spaces
-        # it means the block ended a new one started, so we reset and
-        # continue.
-        if not line.startswith(" " * 8):
-            intent = None
-            continue
-
-        # If we got this far, we are processing receivers for the
-        # activities we are interested in.
-        activity = line.strip().split(" ")[1]
-        package_name = activity.split("/")[0]
-
-        results[intent].append({
-            "package_name": package_name,
-            "activity": activity,
-        })
-
-    return results
-
-
-def parse_dumpsys_battery_daily(output: str) -> list:
-    results = []
-    daily = None
-    daily_updates = []
-    for line in output.splitlines():
-        if line.startswith("  Daily from "):
-            if len(daily_updates) > 0:
-                results.extend(daily_updates)
-                daily_updates = []
-
-            timeframe = line[13:].strip()
-            date_from, date_to = timeframe.strip(":").split(" to ", 1)
-            daily = {"from": date_from[0:10], "to": date_to[0:10]}
-            continue
-
-        if not daily:
-            continue
-
-        if not line.strip().startswith("Update "):
-            continue
-
-        line = line.strip().replace("Update ", "")
-        package_name, vers = line.split(" ", 1)
-        vers_nr = vers.split("=", 1)[1]
-
-        already_seen = False
-        for update in daily_updates:
-            if (package_name == update["package_name"]
-                    and vers_nr == update["vers"]):
-                already_seen = True
-                break
-
-        if not already_seen:
-            daily_updates.append({
-                "action": "update",
-                "from": daily["from"],
-                "to": daily["to"],
-                "package_name": package_name,
-                "vers": vers_nr,
-            })
-
-    if len(daily_updates) > 0:
-        results.extend(daily_updates)
-
-    return results
-
-
-def parse_dumpsys_battery_history(output: str) -> List[Dict[str, Any]]:
-    results = []
-
-    for line in output.splitlines():
-        if line.startswith("Battery History "):
-            continue
-
-        if line.strip() == "":
-            break
-
-        time_elapsed = line.strip().split(" ", 1)[0]
-
-        event = ""
-        if line.find("+job") > 0:
-            event = "start_job"
-            uid = line[line.find("+job")+5:line.find(":")]
-            service = line[line.find(":")+1:].strip('"')
-            package_name = service.split("/")[0]
-        elif line.find("-job") > 0:
-            event = "end_job"
-            uid = line[line.find("-job")+5:line.find(":")]
-            service = line[line.find(":")+1:].strip('"')
-            package_name = service.split("/")[0]
-        elif line.find("+running +wake_lock=") > 0:
-            uid = line[line.find("+running +wake_lock=")+21:line.find(":")]
-            event = "wake"
-            service = line[line.find("*walarm*:")+9:].split(" ")[0].strip('"').strip()
-            if service == "" or "/" not in service:
-                continue
-
-            package_name = service.split("/")[0]
-        elif (line.find("+top=") > 0) or (line.find("-top") > 0):
-            if line.find("+top=") > 0:
-                event = "start_top"
-                top_pos = line.find("+top=")
-            else:
-                event = "end_top"
-                top_pos = line.find("-top=")
-            colon_pos = top_pos+line[top_pos:].find(":")
-            uid = line[top_pos+5:colon_pos]
-            service = ""
-            package_name = line[colon_pos+1:].strip('"')
-        else:
-            continue
-
-        results.append({
-            "time_elapsed": time_elapsed,
-            "event": event,
-            "uid": uid,
-            "package_name": package_name,
-            "service": service,
-        })
-
-    return results
-
-
-def parse_dumpsys_dbinfo(output: str) -> List[Dict[str, Any]]:
-    results = []
-
-    rxp = re.compile(r'.*\[([0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3})\].*\[Pid:\((\d+)\)\](\w+).*sql\=\"(.+?)\"')  # pylint: disable=line-too-long
-    rxp_no_pid = re.compile(r'.*\[([0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3})\][ ]{1}(\w+).*sql\=\"(.+?)\"')  # pylint: disable=line-too-long
-
-    pool = None
-    in_operations = False
-    for line in output.splitlines():
-        if line.startswith("Connection pool for "):
-            pool = line.replace("Connection pool for ", "").rstrip(":")
-
-        if not pool:
-            continue
-
-        if line.strip() == "Most recently executed operations:":
-            in_operations = True
-            continue
-
-        if not in_operations:
-            continue
-
-        if not line.startswith("        "):
-            in_operations = False
-            pool = None
-            continue
-
-        matches = rxp.findall(line)
-        if not matches:
-            matches = rxp_no_pid.findall(line)
-            if not matches:
-                continue
-
-            match = matches[0]
-            results.append({
-                "isodate": match[0],
-                "action": match[1],
-                "sql": match[2],
-                "path": pool,
-            })
-        else:
-            match = matches[0]
-            results.append({
-                "isodate": match[0],
-                "pid": match[1],
-                "action": match[2],
-                "sql": match[3],
-                "path": pool,
-            })
-
-    return results
-
-
-def parse_dumpsys_receiver_resolver_table(output: str) -> Dict[str, Any]:
-    results = {}
-
-    in_receiver_resolver_table = False
-    in_non_data_actions = False
-    intent = None
-    for line in output.splitlines():
-        if line.startswith("Receiver Resolver Table:"):
-            in_receiver_resolver_table = True
-            continue
-
-        if not in_receiver_resolver_table:
-            continue
-
-        if line.startswith("  Non-Data Actions:"):
-            in_non_data_actions = True
-            continue
-
-        if not in_non_data_actions:
-            continue
-
-        # If we hit an empty line, the Non-Data Actions section should be
-        # finished.
-        if line.strip() == "":
-            break
-
-        # We detect the action name.
-        if (line.startswith(" " * 6) and not line.startswith(" " * 8)
-                and ":" in line):
-            intent = line.strip().replace(":", "")
-            results[intent] = []
-            continue
-
-        # If we are not in an intent block yet, skip.
-        if not intent:
-            continue
-
-        # If we are in a block but the line does not start with 8 spaces
-        # it means the block ended a new one started, so we reset and
-        # continue.
-        if not line.startswith(" " * 8):
-            intent = None
-            continue
-
-        # If we got this far, we are processing receivers for the
-        # activities we are interested in.
-        receiver = line.strip().split(" ")[1]
-        package_name = receiver.split("/")[0]
-
-        results[intent].append({
-            "package_name": package_name,
-            "receiver": receiver,
-        })
-
-    return results
-
-
-def parse_dumpsys_appops(output: str) -> List[Dict[str, Any]]:
-    results = []
-    perm = {}
-    package = {}
-    entry = {}
-    uid = None
-    in_packages = False
-
-    for line in output.splitlines():
-        if line.startswith("  Uid 0:"):
-            in_packages = True
-
-        if not in_packages:
-            continue
-
-        if line.startswith("  Uid "):
-            uid = line[6:-1]
-            continue
-
-        if line.startswith("    Package "):
-            if entry:
-                perm["entries"].append(entry)
-                entry = {}
-
-            if package:
-                if perm:
-                    package["permissions"].append(perm)
-
-                perm = {}
-                results.append(package)
-
-            package = {
-                "package_name": line[12:-1],
-                "permissions": [],
-                "uid": uid,
-            }
-            continue
-
-        if line.startswith("      ") and line[6] != " ":
-            if entry:
-                perm["entries"].append(entry)
-                entry = {}
-            if perm:
-                package["permissions"].append(perm)
-                perm = {}
-
-            perm["name"] = line.split()[0]
-            perm["entries"] = []
-            if len(line.split()) > 1:
-                perm["access"] = line.split()[1][1:-2]
-
-            continue
-
-        if line.startswith("          "):
-            # Permission entry like:
-            # Reject: [fg-s]2021-05-19 22:02:52.054 (-314d1h25m2s33ms)
-            if entry:
-                perm["entries"].append(entry)
-                entry = {}
-
-            entry["access"] = line.split(":")[0].strip()
-            entry["type"] = line[line.find("[")+1:line.find("]")]
-
-            try:
-                entry["timestamp"] = convert_datetime_to_iso(
-                    datetime.strptime(
-                        line[line.find("]")+1:line.find("(")].strip(),
-                        "%Y-%m-%d %H:%M:%S.%f"))
-            except ValueError:
-                # Invalid date format
-                pass
-
-        if line.strip() == "":
-            break
-
-    if entry:
-        perm["entries"].append(entry)
-    if perm:
-        package["permissions"].append(perm)
-    if package:
-        results.append(package)
-
-    return results
-
 
 def parse_dumpsys_package_for_details(output: str) -> Dict[str, Any]:
     """
@@ -418,13 +35,11 @@ def parse_dumpsys_package_for_details(output: str) -> Dict[str, Any]:
                 permission = lineinfo[0]
                 granted = None
                 if "granted=" in lineinfo[1]:
-                    granted = ("granted=true" in lineinfo[1])
+                    granted = "granted=true" in lineinfo[1]
 
-                details["permissions"].append({
-                    "name": permission,
-                    "granted": granted,
-                    "type": "install"
-                })
+                details["permissions"].append(
+                    {"name": permission, "granted": granted, "type": "install"}
+                )
 
         if in_runtime_permissions:
             if not line.startswith(" " * 8):
@@ -434,23 +49,18 @@ def parse_dumpsys_package_for_details(output: str) -> Dict[str, Any]:
                 permission = lineinfo[0]
                 granted = None
                 if "granted=" in lineinfo[1]:
-                    granted = ("granted=true" in lineinfo[1])
+                    granted = "granted=true" in lineinfo[1]
 
-                details["permissions"].append({
-                    "name": permission,
-                    "granted": granted,
-                    "type": "runtime"
-                })
+                details["permissions"].append(
+                    {"name": permission, "granted": granted, "type": "runtime"}
+                )
 
         if in_declared_permissions:
             if not line.startswith(" " * 6):
                 in_declared_permissions = False
             else:
                 permission = line.strip().split(":")[0]
-                details["permissions"].append({
-                    "name": permission,
-                    "type": "declared"
-                })
+                details["permissions"].append({"name": permission, "type": "declared"})
         if in_requested_permissions:
             if not line.startswith(" " * 6):
                 in_requested_permissions = False

mvt/android/parsers/getprop.py

@@ -1,29 +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/cmd_check_iocs.py

@@ -8,12 +8,12 @@ import os
 from typing import Optional
 
 from mvt.common.command import Command
+from mvt.common.utils import exec_or_profile
 
 log = logging.getLogger(__name__)
 
 
 class CmdCheckIOCS(Command):
-
     def __init__(
         self,
         target_path: Optional[str] = None,
@@ -21,11 +21,17 @@ class CmdCheckIOCS(Command):
         ioc_files: Optional[list] = None,
         module_name: Optional[str] = None,
         serial: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[dict] = None,
     ) -> None:
-        super().__init__(target_path=target_path, results_path=results_path,
-                         ioc_files=ioc_files, module_name=module_name,
-                         serial=serial, fast_mode=fast_mode, log=log)
+        super().__init__(
+            target_path=target_path,
+            results_path=results_path,
+            ioc_files=ioc_files,
+            module_name=module_name,
+            serial=serial,
+            module_options=module_options,
+            log=log,
+        )
 
         self.name = "check-iocs"
 
@@ -50,22 +56,27 @@ class CmdCheckIOCS(Command):
                 if iocs_module().get_slug() != name_only:
                     continue
 
-                log.info("Loading results from \"%s\" with module %s",
-                         file_name, iocs_module.__name__)
+                log.info(
+                    'Loading results from "%s" with module %s',
+                    file_name,
+                    iocs_module.__name__,
+                )
 
-                m = iocs_module.from_json(file_path,
-                                          log=logging.getLogger(iocs_module.__module__))
+                m = iocs_module.from_json(
+                    file_path, log=logging.getLogger(iocs_module.__module__)
+                )
                 if self.iocs.total_ioc_count > 0:
                     m.indicators = self.iocs
                     m.indicators.log = m.log
 
                 try:
-                    m.check_indicators()
+                    exec_or_profile("m.check_indicators()", globals(), locals())
                 except NotImplementedError:
                     continue
                 else:
                     total_detections += len(m.detected)
 
         if total_detections > 0:
-            log.warning("The check of the results produced %d detections!",
-                        total_detections)
+            log.warning(
+                "The check of the results produced %d detections!", total_detections
+            )

mvt/common/command.py

@@ -12,14 +12,15 @@ from typing import Optional
 
 from mvt.common.indicators import Indicators
 from mvt.common.module import MVTModule, run_module, save_timeline
-from mvt.common.utils import (convert_datetime_to_iso,
-                              generate_hashes_from_path,
-                              get_sha256_from_file_path)
+from mvt.common.utils import (
+    convert_datetime_to_iso,
+    generate_hashes_from_path,
+    get_sha256_from_file_path,
+)
 from mvt.common.version import MVT_VERSION
 
 
 class Command:
-
     def __init__(
         self,
         target_path: Optional[str] = None,
@@ -27,8 +28,8 @@ class Command:
         ioc_files: Optional[list] = None,
         module_name: Optional[str] = None,
         serial: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
-        hashes: Optional[bool] = False,
+        module_options: Optional[dict] = None,
+        hashes: bool = False,
         log: logging.Logger = logging.getLogger(__name__),
     ) -> None:
         self.name = ""
@@ -39,9 +40,13 @@ class Command:
         self.ioc_files = ioc_files if ioc_files else []
         self.module_name = module_name
         self.serial = serial
-        self.fast_mode = fast_mode
         self.log = log
 
+        # This dictionary can contain options that will be passed down from
+        # the Command to all modules. This can for example be used to pass
+        # down a password to decrypt a backup or flags which are need by some modules.
+        self.module_options = module_options if module_options else {}
+
         # This list will contain all executed modules.
         # We can use this to reference e.g. self.executed[0].results.
         self.executed = []
@@ -62,8 +67,9 @@ class Command:
             try:
                 os.makedirs(self.results_path)
             except Exception as exc:
-                self.log.critical("Unable to create output folder %s: %s",
-                                  self.results_path, exc)
+                self.log.critical(
+                    "Unable to create output folder %s: %s", self.results_path, exc
+                )
                 sys.exit(1)
 
     def _setup_logging(self):
@@ -71,10 +77,12 @@ class Command:
             return
 
         logger = logging.getLogger("mvt")
-        file_handler = logging.FileHandler(os.path.join(self.results_path,
-                                                        "command.log"))
-        formatter = logging.Formatter("%(asctime)s - %(name)s - "
-                                      "%(levelname)s - %(message)s")
+        file_handler = logging.FileHandler(
+            os.path.join(self.results_path, "command.log")
+        )
+        formatter = logging.Formatter(
+            "%(asctime)s - %(name)s - " "%(levelname)s - %(message)s"
+        )
         file_handler.setLevel(logging.DEBUG)
         file_handler.setFormatter(formatter)
         logger.addHandler(file_handler)
@@ -84,13 +92,15 @@ class Command:
             return
 
         if len(self.timeline) > 0:
-            save_timeline(self.timeline,
-                          os.path.join(self.results_path, "timeline.csv"))
+            save_timeline(
+                self.timeline, os.path.join(self.results_path, "timeline.csv")
+            )
 
         if len(self.timeline_detected) > 0:
-            save_timeline(self.timeline_detected,
-                          os.path.join(self.results_path,
-                                       "timeline_detected.csv"))
+            save_timeline(
+                self.timeline_detected,
+                os.path.join(self.results_path, "timeline_detected.csv"),
+            )
 
     def _store_info(self) -> None:
         if not self.results_path:
@@ -124,7 +134,7 @@ class Command:
 
         if self.target_path and (os.environ.get("MVT_HASH_FILES") or self.hashes):
             info_hash = get_sha256_from_file_path(info_path)
-            self.log.info("Reference hash of the info.json file: \"%s\"", info_hash)
+            self.log.info('Reference hash of the info.json file: "%s"', info_hash)
 
     def generate_hashes(self) -> None:
         """
@@ -137,8 +147,7 @@ class Command:
             self.hash_values.append(file)
 
     def list_modules(self) -> None:
-        self.log.info("Following is the list of available %s modules:",
-                      self.name)
+        self.log.info("Following is the list of available %s modules:", self.name)
         for module in self.modules:
             self.log.info(" - %s", module.__name__)
 
@@ -152,7 +161,6 @@ class Command:
         raise NotImplementedError
 
     def run(self) -> None:
-
         try:
             self.init()
         except NotImplementedError:
@@ -162,13 +170,15 @@ class Command:
             if self.module_name and module.__name__ != self.module_name:
                 continue
 
-            # FIXME: do we need the logger here
+            # FIXME: do we need the logger here
             module_logger = logging.getLogger(module.__module__)
 
-            m = module(target_path=self.target_path,
-                       results_path=self.results_path,
-                       fast_mode=self.fast_mode,
-                       log=module_logger)
+            m = module(
+                target_path=self.target_path,
+                results_path=self.results_path,
+                module_options=self.module_options,
+                log=module_logger,
+            )
 
             if self.iocs.total_ioc_count:
                 m.indicators = self.iocs

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"
 

mvt/common/indicators.py

@@ -6,8 +6,10 @@
 import json
 import logging
 import os
+from functools import lru_cache
 from typing import Any, Dict, Iterator, List, Optional, Union
 
+import ahocorasick
 from appdirs import user_data_dir
 
 from .url import URL
@@ -34,8 +36,7 @@ class Indicators:
 
         for ioc_file_name in os.listdir(MVT_INDICATORS_FOLDER):
             if ioc_file_name.lower().endswith(".stix2"):
-                self.parse_stix2(os.path.join(MVT_INDICATORS_FOLDER,
-                                              ioc_file_name))
+                self.parse_stix2(os.path.join(MVT_INDICATORS_FOLDER, ioc_file_name))
 
     def _check_stix2_env_variable(self) -> None:
         """
@@ -49,8 +50,9 @@ class Indicators:
             if os.path.isfile(path):
                 self.parse_stix2(path)
             else:
-                self.log.error("Path specified with env MVT_STIX2 is not a valid file: %s",
-                               path)
+                self.log.error(
+                    "Path specified with env MVT_STIX2 is not a valid file: %s", path
+                )
 
     def _new_collection(
         self,
@@ -58,7 +60,7 @@ class Indicators:
         name: Optional[str] = None,
         description: Optional[str] = None,
         file_name: Optional[str] = None,
-        file_path: Optional[str] = None
+        file_path: Optional[str] = None,
     ) -> dict:
         return {
             "id": cid,
@@ -78,8 +80,7 @@ class Indicators:
             "count": 0,
         }
 
-    def _add_indicator(self, ioc: str, ioc_coll: dict,
-                       ioc_coll_list: list) -> None:
+    def _add_indicator(self, ioc: str, ioc_coll: dict, ioc_coll_list: list) -> None:
         ioc = ioc.strip("'")
         if ioc not in ioc_coll_list:
             ioc_coll_list.append(ioc)
@@ -91,43 +92,51 @@ class Indicators:
 
         if key == "domain-name:value":
             # We force domain names to lower case.
-            self._add_indicator(ioc=value.lower(),
-                                ioc_coll=collection,
-                                ioc_coll_list=collection["domains"])
+            self._add_indicator(
+                ioc=value.lower(),
+                ioc_coll=collection,
+                ioc_coll_list=collection["domains"],
+            )
         elif key == "process:name":
-            self._add_indicator(ioc=value,
-                                ioc_coll=collection,
-                                ioc_coll_list=collection["processes"])
+            self._add_indicator(
+                ioc=value, ioc_coll=collection, ioc_coll_list=collection["processes"]
+            )
         elif key == "email-addr:value":
             # We force email addresses to lower case.
-            self._add_indicator(ioc=value.lower(),
-                                ioc_coll=collection,
-                                ioc_coll_list=collection["emails"])
+            self._add_indicator(
+                ioc=value.lower(),
+                ioc_coll=collection,
+                ioc_coll_list=collection["emails"],
+            )
         elif key == "file:name":
-            self._add_indicator(ioc=value,
-                                ioc_coll=collection,
-                                ioc_coll_list=collection["file_names"])
+            self._add_indicator(
+                ioc=value, ioc_coll=collection, ioc_coll_list=collection["file_names"]
+            )
         elif key == "file:path":
-            self._add_indicator(ioc=value,
-                                ioc_coll=collection,
-                                ioc_coll_list=collection["file_paths"])
+            self._add_indicator(
+                ioc=value, ioc_coll=collection, ioc_coll_list=collection["file_paths"]
+            )
         elif key == "file:hashes.sha256":
-            self._add_indicator(ioc=value,
-                                ioc_coll=collection,
-                                ioc_coll_list=collection["files_sha256"])
+            self._add_indicator(
+                ioc=value, ioc_coll=collection, ioc_coll_list=collection["files_sha256"]
+            )
         elif key == "app:id":
-            self._add_indicator(ioc=value,
-                                ioc_coll=collection,
-                                ioc_coll_list=collection["app_ids"])
+            self._add_indicator(
+                ioc=value, ioc_coll=collection, ioc_coll_list=collection["app_ids"]
+            )
         elif key == "configuration-profile:id":
-            self._add_indicator(ioc=value,
-                                ioc_coll=collection,
-                                ioc_coll_list=collection["ios_profile_ids"])
+            self._add_indicator(
+                ioc=value,
+                ioc_coll=collection,
+                ioc_coll_list=collection["ios_profile_ids"],
+            )
 
         elif key == "android-property:name":
-            self._add_indicator(ioc=value,
-                                ioc_coll=collection,
-                                ioc_coll_list=collection["android_property_names"])
+            self._add_indicator(
+                ioc=value,
+                ioc_coll=collection,
+                ioc_coll_list=collection["android_property_names"],
+            )
 
     def parse_stix2(self, file_path: str) -> None:
         """Extract indicators from a STIX2 file.
@@ -142,8 +151,10 @@ class Indicators:
             try:
                 data = json.load(handle)
             except json.decoder.JSONDecodeError:
-                self.log.critical("Unable to parse STIX2 indicator file. "
-                                  "The file is corrupted or in the wrong format!")
+                self.log.critical(
+                    "Unable to parse STIX2 indicator file. "
+                    "The file is corrupted or in the wrong format!"
+                )
                 return
 
         malware = {}
@@ -163,10 +174,13 @@ class Indicators:
 
         collections = []
         for mal_id, mal_values in malware.items():
-            collection = self._new_collection(mal_id, mal_values.get("name"),
-                                              mal_values.get("description"),
-                                              os.path.basename(file_path),
-                                              file_path)
+            collection = self._new_collection(
+                mal_id,
+                mal_values.get("name"),
+                mal_values.get("description"),
+                os.path.basename(file_path),
+                file_path,
+            )
             collections.append(collection)
 
         # We loop through all indicators.
@@ -192,13 +206,17 @@ class Indicators:
                     break
 
         for coll in collections:
-            self.log.info("Extracted %d indicators for collection with name \"%s\"",
-                          coll["count"], coll["name"])
+            self.log.debug(
+                'Extracted %d indicators for collection with name "%s"',
+                coll["count"],
+                coll["name"],
+            )
 
         self.ioc_collections.extend(collections)
 
-    def load_indicators_files(self, files: list,
-                              load_default: Optional[bool] = True) -> None:
+    def load_indicators_files(
+        self, files: list, load_default: Optional[bool] = True
+    ) -> None:
         """
         Load a list of indicators files.
         """
@@ -206,16 +224,14 @@ class Indicators:
             if os.path.isfile(file_path):
                 self.parse_stix2(file_path)
             else:
-                self.log.warning("No indicators file exists at path %s",
-                                 file_path)
+                self.log.warning("No indicators file exists at path %s", file_path)
 
         # Load downloaded indicators and any indicators from env variable.
         if load_default:
             self._load_downloaded_indicators()
 
         self._check_stix2_env_variable()
-        self.log.info("Loaded a total of %d unique indicators",
-                      self.total_ioc_count)
+        self.log.info("Loaded a total of %d unique indicators", self.total_ioc_count)
 
     def get_iocs(self, ioc_type: str) -> Iterator[Dict[str, Any]]:
         for ioc_collection in self.ioc_collections:
@@ -227,6 +243,40 @@ class Indicators:
                     "stix2_file_name": ioc_collection["stix2_file_name"],
                 }
 
+    @lru_cache()
+    def get_ioc_matcher(
+        self, ioc_type: Optional[str] = None, ioc_list: Optional[list] = None
+    ) -> ahocorasick.Automaton:
+        """
+        Build an Aho-Corasick automaton from a list of iocs (i.e indicators)
+        Returns an Aho-Corasick automaton
+
+        This data-structue and algorithim allows for fast matching of a large number
+        of match strings (i.e IOCs) against a large body of text. This will also
+        match strings containing the IOC, so it is important to confirm the
+        match is a valid IOC before using it.
+
+            for _, ioc in domains_automaton.iter(url.domain.lower()):
+                if ioc.value == url.domain.lower():
+                    print(ioc)
+
+        We use an LRU cache to avoid rebuilding the automaton every time we call a
+        function such as check_domain().
+        """
+        automaton = ahocorasick.Automaton()
+        if ioc_type:
+            iocs = self.get_iocs(ioc_type)
+        elif ioc_list:
+            iocs = ioc_list
+        else:
+            raise ValueError("Must provide either ioc_tyxpe or ioc_list")
+
+        for ioc in iocs:
+            automaton.add_word(ioc["value"], ioc)
+        automaton.make_automaton()
+        return automaton
+
+    @lru_cache()
     def check_domain(self, url: str) -> Union[dict, None]:
         """Check if a given URL matches any of the provided domain indicators.
 
@@ -240,6 +290,9 @@ class Indicators:
         if not isinstance(url, str):
             return None
 
+        # Create an Aho-Corasick automaton from the list of domains
+        domain_matcher = self.get_ioc_matcher("domains")
+
         try:
             # First we use the provided URL.
             orig_url = URL(url)
@@ -249,17 +302,20 @@ class Indicators:
                 # HTTP HEAD request.
                 unshortened = orig_url.unshorten()
 
-                self.log.debug("Found a shortened URL %s -> %s",
-                               url, unshortened)
+                self.log.debug("Found a shortened URL %s -> %s", url, unshortened)
                 if unshortened is None:
+                    self.log.warning("Unable to unshorten URL %s", url)
                     return None
 
                 # Now we check for any nested URL shorteners.
                 dest_url = URL(unshortened)
                 if dest_url.check_if_shortened():
-                    self.log.debug("Original URL %s appears to shorten another "
-                                   "shortened URL %s ... checking!",
-                                   orig_url.url, dest_url.url)
+                    self.log.debug(
+                        "Original URL %s appears to shorten another "
+                        "shortened URL %s ... checking!",
+                        orig_url.url,
+                        dest_url.url,
+                    )
                     return self.check_domain(dest_url.url)
 
                 final_url = dest_url
@@ -269,11 +325,15 @@ class Indicators:
         except Exception:
             # If URL parsing failed, we just try to do a simple substring
             # match.
-            for ioc in self.get_iocs("domains"):
+            for idx, ioc in domain_matcher.iter(url):
                 if ioc["value"].lower() in url:
-                    self.log.warning("Maybe found a known suspicious domain %s "
-                                     "matching indicators from \"%s\"",
-                                     url, ioc["name"])
+                    self.log.warning(
+                        "Maybe found a known suspicious domain %s "
+                        'matching indicator "%s" from "%s"',
+                        url,
+                        ioc["value"],
+                        ioc["name"],
+                    )
                     return ioc
 
             # If nothing matched, we can quit here.
@@ -281,31 +341,49 @@ class Indicators:
 
         # If all parsing worked, we start walking through available domain
         # indicators.
-        for ioc in self.get_iocs("domains"):
+        for idx, ioc in domain_matcher.iter(final_url.domain.lower()):
             # First we check the full domain.
             if final_url.domain.lower() == ioc["value"]:
                 if orig_url.is_shortened and orig_url.url != final_url.url:
-                    self.log.warning("Found a known suspicious domain %s "
-                                     "shortened as %s matching indicators from \"%s\"",
-                                     final_url.url, orig_url.url, ioc["name"])
+                    self.log.warning(
+                        "Found a known suspicious domain %s "
+                        'shortened as %s matching indicator "%s" from "%s"',
+                        final_url.url,
+                        orig_url.url,
+                        ioc["value"],
+                        ioc["name"],
+                    )
                 else:
-                    self.log.warning("Found a known suspicious domain %s "
-                                     "matching indicators from \"%s\"",
-                                     final_url.url, ioc["name"])
-
+                    self.log.warning(
+                        "Found a known suspicious domain %s "
+                        'matching indicator "%s" from "%s"',
+                        final_url.url,
+                        ioc["value"],
+                        ioc["name"],
+                    )
                 return ioc
 
-            # Then we just check the top level domain.
+        # Then we just check the top level domain.
+        for idx, ioc in domain_matcher.iter(final_url.top_level.lower()):
             if final_url.top_level.lower() == ioc["value"]:
                 if orig_url.is_shortened and orig_url.url != final_url.url:
-                    self.log.warning("Found a sub-domain with suspicious top "
-                                     "level %s shortened as %s matching "
-                                     "indicators from \"%s\"", final_url.url,
-                                     orig_url.url, ioc["name"])
+                    self.log.warning(
+                        "Found a sub-domain with suspicious top "
+                        "level %s shortened as %s matching "
+                        'indicator "%s" from "%s"',
+                        final_url.url,
+                        orig_url.url,
+                        ioc["value"],
+                        ioc["name"],
+                    )
                 else:
-                    self.log.warning("Found a sub-domain with a suspicious top "
-                                     "level %s matching indicators from \"%s\"",
-                                     final_url.url, ioc["name"])
+                    self.log.warning(
+                        "Found a sub-domain with a suspicious top "
+                        'level %s matching indicator "%s" from "%s"',
+                        final_url.url,
+                        ioc["value"],
+                        ioc["name"],
+                    )
 
                 return ioc
 
@@ -344,16 +422,22 @@ class Indicators:
         proc_name = os.path.basename(process)
         for ioc in self.get_iocs("processes"):
             if proc_name == ioc["value"]:
-                self.log.warning("Found a known suspicious process name \"%s\" "
-                                 "matching indicators from \"%s\"",
-                                 process, ioc["name"])
+                self.log.warning(
+                    'Found a known suspicious process name "%s" '
+                    'matching indicators from "%s"',
+                    process,
+                    ioc["name"],
+                )
                 return ioc
 
             if len(proc_name) == 16:
                 if ioc["value"].startswith(proc_name):
-                    self.log.warning("Found a truncated known suspicious "
-                                     "process name \"%s\" matching indicators from \"%s\"",
-                                     process, ioc["name"])
+                    self.log.warning(
+                        "Found a truncated known suspicious "
+                        'process name "%s" matching indicators from "%s"',
+                        process,
+                        ioc["name"],
+                    )
                     return ioc
 
         return None
@@ -390,9 +474,12 @@ class Indicators:
 
         for ioc in self.get_iocs("emails"):
             if email.lower() == ioc["value"].lower():
-                self.log.warning("Found a known suspicious email address \"%s\" "
-                                 "matching indicators from \"%s\"",
-                                 email, ioc["name"])
+                self.log.warning(
+                    'Found a known suspicious email address "%s" '
+                    'matching indicators from "%s"',
+                    email,
+                    ioc["name"],
+                )
                 return ioc
 
         return None
@@ -411,9 +498,12 @@ class Indicators:
 
         for ioc in self.get_iocs("file_names"):
             if ioc["value"] == file_name:
-                self.log.warning("Found a known suspicious file name \"%s\" "
-                                 "matching indicators from \"%s\"",
-                                 file_name, ioc["name"])
+                self.log.warning(
+                    'Found a known suspicious file name "%s" '
+                    'matching indicators from "%s"',
+                    file_name,
+                    ioc["name"],
+                )
                 return ioc
 
         return None
@@ -439,9 +529,12 @@ class Indicators:
             # Strip any trailing slash from indicator paths to match
             # directories.
             if file_path.startswith(ioc["value"].rstrip("/")):
-                self.log.warning("Found a known suspicious file path \"%s\" "
-                                 "matching indicators form \"%s\"",
-                                 file_path, ioc["name"])
+                self.log.warning(
+                    'Found a known suspicious file path "%s" '
+                    'matching indicators form "%s"',
+                    file_path,
+                    ioc["name"],
+                )
                 return ioc
 
         return None
@@ -462,9 +555,12 @@ class Indicators:
         for ioc in self.get_iocs("processes"):
             parts = file_path.split("/")
             if ioc["value"] in parts:
-                self.log.warning("Found known suspicious process name mentioned in file at "
-                                  "path \"%s\" matching indicators from \"%s\"",
-                                  file_path, ioc["name"])
+                self.log.warning(
+                    "Found known suspicious process name mentioned in file at "
+                    'path "%s" matching indicators from "%s"',
+                    file_path,
+                    ioc["name"],
+                )
                 return ioc
 
         return None
@@ -484,9 +580,12 @@ class Indicators:
 
         for ioc in self.get_iocs("ios_profile_ids"):
             if profile_uuid in ioc["value"]:
-                self.log.warning("Found a known suspicious profile ID \"%s\" "
-                                 "matching indicators from \"%s\"",
-                                 profile_uuid, ioc["name"])
+                self.log.warning(
+                    'Found a known suspicious profile ID "%s" '
+                    'matching indicators from "%s"',
+                    profile_uuid,
+                    ioc["name"],
+                )
                 return ioc
 
         return None
@@ -504,9 +603,12 @@ class Indicators:
 
         for ioc in self.get_iocs("files_sha256"):
             if file_hash.lower() == ioc["value"].lower():
-                self.log.warning("Found a known suspicious file with hash \"%s\" "
-                                 "matching indicators from \"%s\"",
-                                 file_hash, ioc["name"])
+                self.log.warning(
+                    'Found a known suspicious file with hash "%s" '
+                    'matching indicators from "%s"',
+                    file_hash,
+                    ioc["name"],
+                )
                 return ioc
 
         return None
@@ -525,9 +627,12 @@ class Indicators:
 
         for ioc in self.get_iocs("app_ids"):
             if app_id.lower() == ioc["value"].lower():
-                self.log.warning("Found a known suspicious app with ID \"%s\" "
-                                 "matching indicators from \"%s\"", app_id,
-                                 ioc["name"])
+                self.log.warning(
+                    'Found a known suspicious app with ID "%s" '
+                    'matching indicators from "%s"',
+                    app_id,
+                    ioc["name"],
+                )
                 return ioc
 
         return None
@@ -545,9 +650,12 @@ class Indicators:
 
         for ioc in self.get_iocs("android_property_names"):
             if property_name.lower() == ioc["value"].lower():
-                self.log.warning("Found a known suspicious Android property \"%s\" "
-                                 "matching indicators from \"%s\"", property_name,
-                                 ioc["name"])
+                self.log.warning(
+                    'Found a known suspicious Android property "%s" '
+                    'matching indicators from "%s"',
+                    property_name,
+                    ioc["name"],
+                )
                 return ioc
 
         return None

mvt/common/logo.py

@@ -18,8 +18,10 @@ def check_updates() -> None:
         pass
     else:
         if latest_version:
-            rich_print(f"\t\t[bold]Version {latest_version} is available! "
-                       "Upgrade mvt with `pip3 install -U mvt`[/bold]")
+            rich_print(
+                f"\t\t[bold]Version {latest_version} is available! "
+                "Upgrade mvt with `pip3 install -U mvt`[/bold]"
+            )
 
     # Then we check for indicators files updates.
     ioc_updates = IndicatorsUpdates()
@@ -27,8 +29,10 @@ def check_updates() -> None:
     # Before proceeding, we check if we have downloaded an indicators index.
     # If not, there's no point in proceeding with the updates check.
     if ioc_updates.get_latest_update() == 0:
-        rich_print("\t\t[bold]You have not yet downloaded any indicators, check "
-                   "the `download-iocs` command![/bold]")
+        rich_print(
+            "\t\t[bold]You have not yet downloaded any indicators, check "
+            "the `download-iocs` command![/bold]"
+        )
         return
 
     # We only perform this check at a fixed frequency, in order to not
@@ -36,8 +40,10 @@ def check_updates() -> None:
     # multiple times.
     should_check, hours = ioc_updates.should_check()
     if not should_check:
-        rich_print(f"\t\tIndicators updates checked recently, next automatic check "
-                   f"in {int(hours)} hours")
+        rich_print(
+            f"\t\tIndicators updates checked recently, next automatic check "
+            f"in {int(hours)} hours"
+        )
         return
 
     try:
@@ -46,8 +52,10 @@ def check_updates() -> None:
         pass
     else:
         if ioc_to_update:
-            rich_print("\t\t[bold]There are updates to your indicators files! "
-                       "Run the `download-iocs` command to update![/bold]")
+            rich_print(
+                "\t\t[bold]There are updates to your indicators files! "
+                "Run the `download-iocs` command to update![/bold]"
+            )
         else:
             rich_print("\t\tYour indicators files seem to be up to date.")
 

mvt/common/module.py

@@ -11,6 +11,8 @@ from typing import Any, Dict, List, Optional, Union
 
 import simplejson as json
 
+from .utils import exec_or_profile
+
 
 class DatabaseNotFoundError(Exception):
     pass
@@ -35,9 +37,9 @@ class MVTModule:
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        module_options: Optional[Dict[str, Any]] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: Union[List[Dict[str, Any]], Dict[str, Any], None] = None
+        results: Union[List[Dict[str, Any]], Dict[str, Any], None] = None,
     ) -> None:
         """Initialize module.
 
@@ -57,7 +59,7 @@ class MVTModule:
         self.file_path = file_path
         self.target_path = target_path
         self.results_path = results_path
-        self.fast_mode = fast_mode
+        self.module_options = module_options if module_options else {}
         self.log = log
         self.indicators = None
         self.results = results if results else []
@@ -70,8 +72,7 @@ class MVTModule:
         with open(json_path, "r", encoding="utf-8") as handle:
             results = json.load(handle)
             if log:
-                log.info("Loaded %d results from \"%s\"",
-                         len(results), json_path)
+                log.info('Loaded %d results from "%s"', len(results), json_path)
             return cls(results=results, log=log)
 
     def get_slug(self) -> str:
@@ -99,20 +100,21 @@ class MVTModule:
 
         if self.results:
             results_file_name = f"{name}.json"
-            results_json_path = os.path.join(self.results_path,
-                                             results_file_name)
+            results_json_path = os.path.join(self.results_path, results_file_name)
             with open(results_json_path, "w", encoding="utf-8") as handle:
                 try:
                     json.dump(self.results, handle, indent=4, default=str)
                 except Exception as exc:
-                    self.log.error("Unable to store results of module %s to file %s: %s",
-                                   self.__class__.__name__, results_file_name,
-                                   exc)
+                    self.log.error(
+                        "Unable to store results of module %s to file %s: %s",
+                        self.__class__.__name__,
+                        results_file_name,
+                        exc,
+                    )
 
         if self.detected:
             detected_file_name = f"{name}_detected.json"
-            detected_json_path = os.path.join(self.results_path,
-                                              detected_file_name)
+            detected_json_path = os.path.join(self.results_path, detected_file_name)
             with open(detected_json_path, "w", encoding="utf-8") as handle:
                 json.dump(self.detected, handle, indent=4, default=str)
 
@@ -151,8 +153,7 @@ class MVTModule:
 
         # De-duplicate timeline entries.
         self.timeline = self._deduplicate_timeline(self.timeline)
-        self.timeline_detected = self._deduplicate_timeline(
-            self.timeline_detected)
+        self.timeline_detected = self._deduplicate_timeline(self.timeline_detected)
 
     def run(self) -> None:
         """Run the main module procedure."""
@@ -163,44 +164,65 @@ def run_module(module: MVTModule) -> None:
     module.log.info("Running module %s...", module.__class__.__name__)
 
     try:
-        module.run()
+        exec_or_profile("module.run()", globals(), locals())
     except NotImplementedError:
-        module.log.exception("The run() procedure of module %s was not implemented yet!",
-                             module.__class__.__name__)
+        module.log.exception(
+            "The run() procedure of module %s was not implemented yet!",
+            module.__class__.__name__,
+        )
     except InsufficientPrivileges as exc:
-        module.log.info("Insufficient privileges for module %s: %s",
-                        module.__class__.__name__, exc)
+        module.log.info(
+            "Insufficient privileges for module %s: %s", module.__class__.__name__, exc
+        )
     except DatabaseNotFoundError as exc:
-        module.log.info("There might be no data to extract by module %s: %s",
-                        module.__class__.__name__, exc)
+        module.log.info(
+            "There might be no data to extract by module %s: %s",
+            module.__class__.__name__,
+            exc,
+        )
     except DatabaseCorruptedError as exc:
-        module.log.error("The %s module database seems to be corrupted: %s",
-                         module.__class__.__name__, exc)
+        module.log.error(
+            "The %s module database seems to be corrupted: %s",
+            module.__class__.__name__,
+            exc,
+        )
     except Exception as exc:
-        module.log.exception("Error in running extraction from module %s: %s",
-                             module.__class__.__name__, exc)
+        module.log.exception(
+            "Error in running extraction from module %s: %s",
+            module.__class__.__name__,
+            exc,
+        )
     else:
         try:
-            module.check_indicators()
+            exec_or_profile("module.check_indicators()", globals(), locals())
         except NotImplementedError:
-            module.log.info("The %s module does not support checking for indicators",
-                            module.__class__.__name__)
+            module.log.info(
+                "The %s module does not support checking for indicators",
+                module.__class__.__name__,
+            )
         except Exception as exc:
-            module.log.exception("Error when checking indicators from module %s: %s",
-                                 module.__class__.__name__, exc)
+            module.log.exception(
+                "Error when checking indicators from module %s: %s",
+                module.__class__.__name__,
+                exc,
+            )
 
         else:
             if module.indicators and not module.detected:
-                module.log.info("The %s module produced no detections!",
-                                module.__class__.__name__)
+                module.log.info(
+                    "The %s module produced no detections!", module.__class__.__name__
+                )
 
         try:
             module.to_timeline()
         except NotImplementedError:
             pass
         except Exception as exc:
-            module.log.exception("Error when serializing data from module %s: %s",
-                                 module.__class__.__name__, exc)
+            module.log.exception(
+                "Error when serializing data from module %s: %s",
+                module.__class__.__name__,
+                exc,
+            )
 
         module.save_to_json()
 
@@ -213,15 +235,19 @@ def save_timeline(timeline: list, timeline_path: str) -> None:
 
     """
     with open(timeline_path, "a+", encoding="utf-8") as handle:
-        csvoutput = csv.writer(handle, delimiter=",", quotechar="\"",
-                               quoting=csv.QUOTE_ALL, escapechar='\\')
+        csvoutput = csv.writer(
+            handle, delimiter=",", quotechar='"', quoting=csv.QUOTE_ALL, escapechar="\\"
+        )
         csvoutput.writerow(["UTC Timestamp", "Plugin", "Event", "Description"])
 
-        for event in sorted(timeline, key=lambda x: x["timestamp"]
-                            if x["timestamp"] is not None else ""):
-            csvoutput.writerow([
-                event.get("timestamp"),
-                event.get("module"),
-                event.get("event"),
-                event.get("data"),
-            ])
+        for event in sorted(
+            timeline, key=lambda x: x["timestamp"] if x["timestamp"] is not None else ""
+        ):
+            csvoutput.writerow(
+                [
+                    event.get("timestamp"),
+                    event.get("module"),
+                    event.get("event"),
+                    event.get("data"),
+                ]
+            )

mvt/common/options.py

@@ -16,8 +16,10 @@ class MutuallyExclusiveOption(Option):
         help_msg = kwargs.get("help", "")
         if self.mutually_exclusive:
             ex_str = ", ".join(self.mutually_exclusive)
-            kwargs["help"] = (f"{help_msg} NOTE: This argument is mutually exclusive with arguments"
-                              f"[{ex_str}].")
+            kwargs["help"] = (
+                f"{help_msg} NOTE: This argument is mutually exclusive with arguments"
+                f"[{ex_str}]."
+            )
 
         super().__init__(*args, **kwargs)
 

mvt/common/updates.py

@@ -22,7 +22,6 @@ INDICATORS_CHECK_FREQUENCY = 12
 
 
 class MVTUpdates:
-
     def check(self) -> str:
         res = requests.get("https://pypi.org/pypi/mvt/json")
         data = res.json()
@@ -35,7 +34,6 @@ class MVTUpdates:
 
 
 class IndicatorsUpdates:
-
     def __init__(self) -> None:
         self.github_raw_url = "https://raw.githubusercontent.com/{}/{}/{}/{}"
 
@@ -47,10 +45,12 @@ class IndicatorsUpdates:
         if not os.path.exists(MVT_DATA_FOLDER):
             os.makedirs(MVT_DATA_FOLDER)
 
-        self.latest_update_path = os.path.join(MVT_DATA_FOLDER,
-                                               "latest_indicators_update")
-        self.latest_check_path = os.path.join(MVT_DATA_FOLDER,
-                                              "latest_indicators_check")
+        self.latest_update_path = os.path.join(
+            MVT_DATA_FOLDER, "latest_indicators_update"
+        )
+        self.latest_check_path = os.path.join(
+            MVT_DATA_FOLDER, "latest_indicators_check"
+        )
 
     def get_latest_check(self) -> int:
         if not os.path.exists(self.latest_check_path):
@@ -85,12 +85,16 @@ class IndicatorsUpdates:
             handle.write(str(timestamp))
 
     def get_remote_index(self) -> Optional[dict]:
-        url = self.github_raw_url.format(self.index_owner, self.index_repo,
-                                         self.index_branch, self.index_path)
+        url = self.github_raw_url.format(
+            self.index_owner, self.index_repo, self.index_branch, self.index_path
+        )
         res = requests.get(url)
         if res.status_code != 200:
-            log.error("Failed to retrieve indicators index located at %s (error %d)",
-                      url, res.status_code)
+            log.error(
+                "Failed to retrieve indicators index located at %s (error %d)",
+                url,
+                res.status_code,
+            )
             return None
 
         return yaml.safe_load(res.content)
@@ -98,8 +102,11 @@ class IndicatorsUpdates:
     def download_remote_ioc(self, ioc_url: str) -> Optional[str]:
         res = requests.get(ioc_url)
         if res.status_code != 200:
-            log.error("Failed to download indicators file from %s (error %d)",
-                      ioc_url, res.status_code)
+            log.error(
+                "Failed to download indicators file from %s (error %d)",
+                ioc_url,
+                res.status_code,
+            )
             return None
 
         clean_file_name = ioc_url.lstrip("https://").replace("/", "_")
@@ -135,28 +142,37 @@ class IndicatorsUpdates:
                 ioc_url = ioc.get("download_url", "")
 
             if not ioc_url:
-                log.error("Could not find a way to download indicator file for %s",
-                          ioc.get("name"))
+                log.error(
+                    "Could not find a way to download indicator file for %s",
+                    ioc.get("name"),
+                )
                 continue
 
             ioc_local_path = self.download_remote_ioc(ioc_url)
             if not ioc_local_path:
                 continue
 
-            log.info("Downloaded indicators \"%s\" to %s",
-                     ioc.get("name"), ioc_local_path)
+            log.info(
+                'Downloaded indicators "%s" to %s', ioc.get("name"), ioc_local_path
+            )
 
         self.set_latest_update()
 
-    def _get_remote_file_latest_commit(self, owner: str, repo: str,
-                                       branch: str, path: str) -> int:
+    def _get_remote_file_latest_commit(
+        self, owner: str, repo: str, branch: str, path: str
+    ) -> int:
         # TODO: The branch is currently not taken into consideration.
         #       How do we specify which branch to look up to the API?
-        file_commit_url = f"https://api.github.com/repos/{owner}/{repo}/commits?path={path}"
+        file_commit_url = (
+            f"https://api.github.com/repos/{owner}/{repo}/commits?path={path}"
+        )
         res = requests.get(file_commit_url)
         if res.status_code != 200:
-            log.error("Failed to get details about file %s (error %d)",
-                      file_commit_url, res.status_code)
+            log.error(
+                "Failed to get details about file %s (error %d)",
+                file_commit_url,
+                res.status_code,
+            )
             return -1
 
         details = res.json()
@@ -164,13 +180,16 @@ class IndicatorsUpdates:
             return -1
 
         latest_commit = details[0]
-        latest_commit_date = latest_commit.get("commit", {}).get("author", {}).get("date", None)
+        latest_commit_date = (
+            latest_commit.get("commit", {}).get("author", {}).get("date", None)
+        )
         if not latest_commit_date:
-            log.error("Failed to retrieve date of latest update to indicators index file")
+            log.error(
+                "Failed to retrieve date of latest update to indicators index file"
+            )
             return -1
 
-        latest_commit_dt = datetime.strptime(latest_commit_date,
-                                             '%Y-%m-%dT%H:%M:%SZ')
+        latest_commit_dt = datetime.strptime(latest_commit_date, "%Y-%m-%dT%H:%M:%SZ")
         latest_commit_ts = int(latest_commit_dt.timestamp())
 
         return latest_commit_ts
@@ -192,10 +211,9 @@ class IndicatorsUpdates:
         self.set_latest_check()
 
         latest_update = self.get_latest_update()
-        latest_commit_ts = self._get_remote_file_latest_commit(self.index_owner,
-                                                               self.index_repo,
-                                                               self.index_branch,
-                                                               self.index_path)
+        latest_commit_ts = self._get_remote_file_latest_commit(
+            self.index_owner, self.index_repo, self.index_branch, self.index_path
+        )
 
         if latest_update < latest_commit_ts:
             return True
@@ -214,10 +232,9 @@ class IndicatorsUpdates:
             branch = github.get("branch", "main")
             path = github.get("path", "")
 
-            file_latest_commit_ts = self._get_remote_file_latest_commit(owner,
-                                                                        repo,
-                                                                        branch,
-                                                                        path)
+            file_latest_commit_ts = self._get_remote_file_latest_commit(
+                owner, repo, branch, path
+            )
             if latest_update < file_latest_commit_ts:
                 return True
 

mvt/common/url.py

@@ -254,7 +254,6 @@ SHORTENER_DOMAINS = [
 
 
 class URL:
-
     def __init__(self, url: str) -> None:
         if isinstance(url, bytes):
             url = url.decode()
@@ -273,9 +272,11 @@ class URL:
         :rtype: str
 
         """
-        return get_tld(self.url,
-                       as_object=True,
-                       fix_protocol=True).parsed_url.netloc.lower().lstrip("www.")
+        return (
+            get_tld(self.url, as_object=True, fix_protocol=True)
+            .parsed_url.netloc.lower()
+            .lstrip("www.")
+        )
 
     def get_top_level(self) -> str:
         """Get only the top-level domain from a URL.
@@ -286,9 +287,7 @@ class URL:
         :rtype: str
 
         """
-        return get_tld(self.url,
-                       as_object=True,
-                       fix_protocol=True).fld.lower()
+        return get_tld(self.url, as_object=True, fix_protocol=True).fld.lower()
 
     def check_if_shortened(self) -> bool:
         """Check if the URL is among list of shortener services.

mvt/common/utils.py

@@ -3,6 +3,7 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
+import cProfile
 import datetime
 import hashlib
 import logging
@@ -42,7 +43,7 @@ def convert_datetime_to_iso(date_time: datetime.datetime) -> str:
 
 
 def convert_unix_to_utc_datetime(
-        timestamp: Union[int, float, str]
+    timestamp: Union[int, float, str]
 ) -> datetime.datetime:
     """Converts a unix epoch timestamp to UTC datetime.
 
@@ -69,8 +70,7 @@ def convert_unix_to_iso(timestamp: Union[int, float, str]) -> str:
         return ""
 
 
-def convert_mactime_to_datetime(timestamp: Union[int, float],
-                                from_2001: bool = True):
+def convert_mactime_to_datetime(timestamp: Union[int, float], from_2001: bool = True):
     """Converts Mac Standard Time to a datetime.
 
     :param timestamp: MacTime timestamp (either int or float).
@@ -111,8 +111,7 @@ def convert_mactime_to_iso(timestamp: int, from_2001: bool = True):
 
     """
 
-    return convert_datetime_to_iso(
-        convert_mactime_to_datetime(timestamp, from_2001))
+    return convert_datetime_to_iso(convert_mactime_to_datetime(timestamp, from_2001))
 
 
 def check_for_links(text: str) -> list:
@@ -185,18 +184,20 @@ def generate_hashes_from_path(path: str, log) -> Iterator[dict]:
         hash_value = get_sha256_from_file_path(path)
         yield {"file_path": path, "sha256": hash_value}
     elif os.path.isdir(path):
-        for (root, _, files) in os.walk(path):
+        for root, _, files in os.walk(path):
             for file in files:
                 file_path = os.path.join(root, file)
                 try:
                     sha256 = get_sha256_from_file_path(file_path)
                 except FileNotFoundError:
-                    log.error("Failed to hash the file %s: might be a symlink",
-                              file_path)
+                    log.error(
+                        "Failed to hash the file %s: might be a symlink", file_path
+                    )
                     continue
                 except PermissionError:
-                    log.error("Failed to hash the file %s: permission denied",
-                              file_path)
+                    log.error(
+                        "Failed to hash the file %s: permission denied", file_path
+                    )
                     continue
 
                 yield {"file_path": file_path, "sha256": sha256}
@@ -225,3 +226,11 @@ def set_verbose_logging(verbose: bool = False):
         handler.setLevel(logging.DEBUG)
     else:
         handler.setLevel(logging.INFO)
+
+
+def exec_or_profile(module, globals, locals):
+    """Hook for profiling MVT modules"""
+    if int(os.environ.get("MVT_PROFILE", False)):
+        cProfile.runctx(module, globals, locals)
+    else:
+        exec(module, globals, locals)

mvt/common/version.py

@@ -3,4 +3,4 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
-MVT_VERSION = "2.2.6"
+MVT_VERSION = "2.4.2"

mvt/common/virustotal.py

@@ -23,17 +23,20 @@ class VTQuotaExceeded(Exception):
 
 def virustotal_lookup(file_hash: str):
     if MVT_VT_API_KEY not in os.environ:
-        raise VTNoKey("No VirusTotal API key provided: to use VirusTotal "
-                      "lookups please provide your API key with "
-                      "`export MVT_VT_API_KEY=<key>`")
+        raise VTNoKey(
+            "No VirusTotal API key provided: to use VirusTotal "
+            "lookups please provide your API key with "
+            "`export MVT_VT_API_KEY=<key>`"
+        )
 
     headers = {
         "User-Agent": "VirusTotal",
         "Content-Type": "application/json",
         "x-apikey": os.environ[MVT_VT_API_KEY],
     }
-    res = requests.get(f"https://www.virustotal.com/api/v3/files/{file_hash}",
-                       headers=headers)
+    res = requests.get(
+        f"https://www.virustotal.com/api/v3/files/{file_hash}", headers=headers
+    )
 
     if res.status_code == 200:
         report = res.json()

mvt/ios/cli.py

@@ -11,14 +11,23 @@ import click
 from rich.prompt import Prompt
 
 from mvt.common.cmd_check_iocs import CmdCheckIOCS
-from mvt.common.help import (HELP_MSG_FAST, HELP_MSG_HASHES, HELP_MSG_IOC,
-                             HELP_MSG_LIST_MODULES, HELP_MSG_MODULE,
-                             HELP_MSG_OUTPUT, HELP_MSG_VERBOSE)
+from mvt.common.help import (
+    HELP_MSG_FAST,
+    HELP_MSG_HASHES,
+    HELP_MSG_IOC,
+    HELP_MSG_LIST_MODULES,
+    HELP_MSG_MODULE,
+    HELP_MSG_OUTPUT,
+    HELP_MSG_VERBOSE,
+)
 from mvt.common.logo import logo
 from mvt.common.options import MutuallyExclusiveOption
 from mvt.common.updates import IndicatorsUpdates
-from mvt.common.utils import (generate_hashes_from_path, init_logging,
-                              set_verbose_logging)
+from mvt.common.utils import (
+    generate_hashes_from_path,
+    init_logging,
+    set_verbose_logging,
+)
 
 from .cmd_check_backup import CmdIOSCheckBackup
 from .cmd_check_fs import CmdIOSCheckFS
@@ -32,41 +41,55 @@ log = logging.getLogger("mvt")
 
 # Set this environment variable to a password if needed.
 MVT_IOS_BACKUP_PASSWORD = "MVT_IOS_BACKUP_PASSWORD"
-CONTEXT_SETTINGS = dict(help_option_names=['-h', '--help'])
+CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])
 
 
-#==============================================================================
+# ==============================================================================
 # Main
-#==============================================================================
+# ==============================================================================
 @click.group(invoke_without_command=False)
 def cli():
     logo()
 
 
-#==============================================================================
+# ==============================================================================
 # Command: version
-#==============================================================================
+# ==============================================================================
 @cli.command("version", help="Show the currently installed version of MVT")
 def version():
     return
 
 
-#==============================================================================
+# ==============================================================================
 # Command: decrypt-backup
-#==============================================================================
-@cli.command("decrypt-backup", help="Decrypt an encrypted iTunes backup",
-             context_settings=CONTEXT_SETTINGS)
-@click.option("--destination", "-d", required=True,
-              help="Path to the folder where to store the decrypted backup")
-@click.option("--password", "-p", cls=MutuallyExclusiveOption,
-              help="Password to use to decrypt the backup (or, set "
-                   f"{MVT_IOS_BACKUP_PASSWORD} environment variable)",
-              mutually_exclusive=["key_file"])
-@click.option("--key-file", "-k", cls=MutuallyExclusiveOption,
-              type=click.Path(exists=True),
-              help="File containing raw encryption key to use to decrypt "
-                   "the backup",
-              mutually_exclusive=["password"])
+# ==============================================================================
+@cli.command(
+    "decrypt-backup",
+    help="Decrypt an encrypted iTunes backup",
+    context_settings=CONTEXT_SETTINGS,
+)
+@click.option(
+    "--destination",
+    "-d",
+    required=True,
+    help="Path to the folder where to store the decrypted backup",
+)
+@click.option(
+    "--password",
+    "-p",
+    cls=MutuallyExclusiveOption,
+    help="Password to use to decrypt the backup (or, set "
+    f"{MVT_IOS_BACKUP_PASSWORD} environment variable)",
+    mutually_exclusive=["key_file"],
+)
+@click.option(
+    "--key-file",
+    "-k",
+    cls=MutuallyExclusiveOption,
+    type=click.Path(exists=True),
+    help="File containing raw encryption key to use to decrypt " "the backup",
+    mutually_exclusive=["password"],
+)
 @click.option("--hashes", "-H", is_flag=True, help=HELP_MSG_HASHES)
 @click.argument("BACKUP_PATH", type=click.Path(exists=True))
 @click.pass_context
@@ -75,22 +98,28 @@ def decrypt_backup(ctx, destination, password, key_file, hashes, backup_path):
 
     if key_file:
         if MVT_IOS_BACKUP_PASSWORD in os.environ:
-            log.info("Ignoring %s environment variable, using --key-file"
-                     "'%s' instead", MVT_IOS_BACKUP_PASSWORD, key_file)
+            log.info(
+                "Ignoring %s environment variable, using --key-file" "'%s' instead",
+                MVT_IOS_BACKUP_PASSWORD,
+                key_file,
+            )
 
         backup.decrypt_with_key_file(key_file)
     elif password:
-        log.info("Your password may be visible in the process table because it "
-                 "was supplied on the command line!")
+        log.info(
+            "Your password may be visible in the process table because it "
+            "was supplied on the command line!"
+        )
 
         if MVT_IOS_BACKUP_PASSWORD in os.environ:
-            log.info("Ignoring %s environment variable, using --password"
-                     "argument instead", MVT_IOS_BACKUP_PASSWORD)
+            log.info(
+                "Ignoring %s environment variable, using --password" "argument instead",
+                MVT_IOS_BACKUP_PASSWORD,
+            )
 
         backup.decrypt_with_password(password)
     elif MVT_IOS_BACKUP_PASSWORD in os.environ:
-        log.info("Using password from %s environment variable",
-                  MVT_IOS_BACKUP_PASSWORD)
+        log.info("Using password from %s environment variable", MVT_IOS_BACKUP_PASSWORD)
         backup.decrypt_with_password(os.environ[MVT_IOS_BACKUP_PASSWORD])
     else:
         sekrit = Prompt.ask("Enter backup password", password=True)
@@ -112,33 +141,45 @@ def decrypt_backup(ctx, destination, password, key_file, hashes, backup_path):
             json.dump(info, handle, indent=4)
 
 
-#==============================================================================
+# ==============================================================================
 # Command: extract-key
-#==============================================================================
-@cli.command("extract-key", help="Extract decryption key from an iTunes backup",
-             context_settings=CONTEXT_SETTINGS)
-@click.option("--password", "-p",
-              help="Password to use to decrypt the backup (or, set "
-                   f"{MVT_IOS_BACKUP_PASSWORD} environment variable)")
-@click.option("--key-file", "-k",
-              help="Key file to be written (if unset, will print to STDOUT)",
-              required=False,
-              type=click.Path(exists=False, file_okay=True, dir_okay=False,
-                              writable=True))
+# ==============================================================================
+@cli.command(
+    "extract-key",
+    help="Extract decryption key from an iTunes backup",
+    context_settings=CONTEXT_SETTINGS,
+)
+@click.option(
+    "--password",
+    "-p",
+    help="Password to use to decrypt the backup (or, set "
+    f"{MVT_IOS_BACKUP_PASSWORD} environment variable)",
+)
+@click.option(
+    "--key-file",
+    "-k",
+    help="Key file to be written (if unset, will print to STDOUT)",
+    required=False,
+    type=click.Path(exists=False, file_okay=True, dir_okay=False, writable=True),
+)
 @click.argument("BACKUP_PATH", type=click.Path(exists=True))
 def extract_key(password, key_file, backup_path):
     backup = DecryptBackup(backup_path)
 
     if password:
-        log.info("Your password may be visible in the process table because it "
-                 "was supplied on the command line!")
+        log.info(
+            "Your password may be visible in the process table because it "
+            "was supplied on the command line!"
+        )
 
         if MVT_IOS_BACKUP_PASSWORD in os.environ:
-            log.info("Ignoring %s environment variable, using --password "
-                     "argument instead", MVT_IOS_BACKUP_PASSWORD)
+            log.info(
+                "Ignoring %s environment variable, using --password "
+                "argument instead",
+                MVT_IOS_BACKUP_PASSWORD,
+            )
     elif MVT_IOS_BACKUP_PASSWORD in os.environ:
-        log.info("Using password from %s environment variable",
-                 MVT_IOS_BACKUP_PASSWORD)
+        log.info("Using password from %s environment variable", MVT_IOS_BACKUP_PASSWORD)
         password = os.environ[MVT_IOS_BACKUP_PASSWORD]
     else:
         password = Prompt.ask("Enter backup password", password=True)
@@ -150,15 +191,23 @@ def extract_key(password, key_file, backup_path):
         backup.write_key(key_file)
 
 
-#==============================================================================
+# ==============================================================================
 # Command: check-backup
-#==============================================================================
-@cli.command("check-backup", help="Extract artifacts from an iTunes backup",
-             context_settings=CONTEXT_SETTINGS)
-@click.option("--iocs", "-i", type=click.Path(exists=True), multiple=True,
-              default=[], help=HELP_MSG_IOC)
-@click.option("--output", "-o", type=click.Path(exists=False),
-              help=HELP_MSG_OUTPUT)
+# ==============================================================================
+@cli.command(
+    "check-backup",
+    help="Extract artifacts from an iTunes backup",
+    context_settings=CONTEXT_SETTINGS,
+)
+@click.option(
+    "--iocs",
+    "-i",
+    type=click.Path(exists=True),
+    multiple=True,
+    default=[],
+    help=HELP_MSG_IOC,
+)
+@click.option("--output", "-o", type=click.Path(exists=False), help=HELP_MSG_OUTPUT)
 @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)
@@ -166,12 +215,20 @@ def extract_key(password, key_file, backup_path):
 @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, fast, list_modules, module, hashes, verbose, backup_path):
+def check_backup(
+    ctx, iocs, output, fast, list_modules, module, hashes, verbose, backup_path
+):
     set_verbose_logging(verbose)
+    module_options = {"fast_mode": fast}
 
-    cmd = CmdIOSCheckBackup(target_path=backup_path, results_path=output,
-                            ioc_files=iocs, module_name=module, fast_mode=fast,
-                            hashes=hashes)
+    cmd = CmdIOSCheckBackup(
+        target_path=backup_path,
+        results_path=output,
+        ioc_files=iocs,
+        module_name=module,
+        module_options=module_options,
+        hashes=hashes,
+    )
 
     if list_modules:
         cmd.list_modules()
@@ -182,19 +239,28 @@ def check_backup(ctx, iocs, output, fast, list_modules, module, hashes, verbose,
     cmd.run()
 
     if cmd.detected_count > 0:
-        log.warning("The analysis of the backup produced %d detections!",
-                    cmd.detected_count)
+        log.warning(
+            "The analysis of the backup produced %d detections!", cmd.detected_count
+        )
 
 
-#==============================================================================
+# ==============================================================================
 # Command: check-fs
-#==============================================================================
-@cli.command("check-fs", help="Extract artifacts from a full filesystem dump",
-             context_settings=CONTEXT_SETTINGS)
-@click.option("--iocs", "-i", type=click.Path(exists=True), multiple=True,
-              default=[], help=HELP_MSG_IOC)
-@click.option("--output", "-o", type=click.Path(exists=False),
-              help=HELP_MSG_OUTPUT)
+# ==============================================================================
+@cli.command(
+    "check-fs",
+    help="Extract artifacts from a full filesystem dump",
+    context_settings=CONTEXT_SETTINGS,
+)
+@click.option(
+    "--iocs",
+    "-i",
+    type=click.Path(exists=True),
+    multiple=True,
+    default=[],
+    help=HELP_MSG_IOC,
+)
+@click.option("--output", "-o", type=click.Path(exists=False), help=HELP_MSG_OUTPUT)
 @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)
@@ -204,9 +270,16 @@ def check_backup(ctx, iocs, output, fast, list_modules, module, hashes, verbose,
 @click.pass_context
 def check_fs(ctx, iocs, output, fast, list_modules, module, hashes, verbose, dump_path):
     set_verbose_logging(verbose)
-    cmd = CmdIOSCheckFS(target_path=dump_path, results_path=output,
-                        ioc_files=iocs, module_name=module, fast_mode=fast,
-                        hashes=hashes)
+    module_options = {"fast_mode": fast}
+
+    cmd = CmdIOSCheckFS(
+        target_path=dump_path,
+        results_path=output,
+        ioc_files=iocs,
+        module_name=module,
+        module_options=module_options,
+        hashes=hashes,
+    )
 
     if list_modules:
         cmd.list_modules()
@@ -217,17 +290,28 @@ def check_fs(ctx, iocs, output, fast, list_modules, module, hashes, verbose, dum
     cmd.run()
 
     if cmd.detected_count > 0:
-        log.warning("The analysis of the iOS filesystem produced %d detections!",
-                    cmd.detected_count)
+        log.warning(
+            "The analysis of the iOS filesystem produced %d detections!",
+            cmd.detected_count,
+        )
 
 
-#==============================================================================
+# ==============================================================================
 # Command: check-iocs
-#==============================================================================
-@cli.command("check-iocs", help="Compare stored JSON results to provided indicators",
-             context_settings=CONTEXT_SETTINGS)
-@click.option("--iocs", "-i", type=click.Path(exists=True), multiple=True,
-              default=[], help=HELP_MSG_IOC)
+# ==============================================================================
+@cli.command(
+    "check-iocs",
+    help="Compare stored JSON results to provided indicators",
+    context_settings=CONTEXT_SETTINGS,
+)
+@click.option(
+    "--iocs",
+    "-i",
+    type=click.Path(exists=True),
+    multiple=True,
+    default=[],
+    help=HELP_MSG_IOC,
+)
 @click.option("--list-modules", "-l", is_flag=True, help=HELP_MSG_LIST_MODULES)
 @click.option("--module", "-m", help=HELP_MSG_MODULE)
 @click.argument("FOLDER", type=click.Path(exists=True))
@@ -243,11 +327,14 @@ def check_iocs(ctx, iocs, list_modules, module, folder):
     cmd.run()
 
 
-#==============================================================================
+# ==============================================================================
 # Command: download-iocs
-#==============================================================================
-@cli.command("download-iocs", help="Download public STIX2 indicators",
-             context_settings=CONTEXT_SETTINGS)
+# ==============================================================================
+@cli.command(
+    "download-iocs",
+    help="Download public STIX2 indicators",
+    context_settings=CONTEXT_SETTINGS,
+)
 def download_iocs():
     ioc_updates = IndicatorsUpdates()
     ioc_updates.update()