Bump version number

Optimise domain checking performance

.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/flake8.yml

@@ -1,26 +0,0 @@
-name: Flake8
-
-on:
-  push:
-    paths:
-      - '*.py'
-
-jobs:
-  flake8_py3:
-    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 flake8
-        run: pip install flake8
-      - name: Run flake8
-        uses: suo/flake8-github-action@releases/v1
-        with:
-          checkName: 'flake8_py3'   # NOTE: this needs to be the same as the job name
-        env:
-          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/python-package.yml

@@ -16,8 +16,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        # python-version: [3.7, 3.8, 3.9]
-        python-version: [3.8, 3.9]
+        python-version: ['3.8', '3.9', '3.10']
 
     steps:
     - uses: actions/checkout@v2

.github/workflows/ruff.yml

@@ -0,0 +1,21 @@
+name: Ruff
+on: [push]
+
+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
+      - name: ruff
+        run: |
+          ruff check .

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

@@ -0,0 +1,89 @@
+"""
+Python script to download the Apple RSS feed and parse it.
+"""
+
+import json
+import os
+import urllib.request
+from xml.dom.minidom import parseString
+
+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")
+    print("Downloaded RSS feed from Apple.")
+    return rss_feed
+
+
+def parse_latest_ios_versions(rss_feed_text):
+    latest_ios_versions = []
+
+    parsed_feed = parseString(rss_feed_text)
+    for item in parsed_feed.getElementsByTagName("item"):
+        title = item.getElementsByTagName("title")[0].firstChild.data
+        if not title.startswith("iOS"):
+            continue
+
+        import re
+
+        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
+
+        release_info = build_match.groupdict()
+        if release_info["beta"]:
+            print("Skipping beta release:", title)
+            continue
+
+        release_info.pop("beta")
+        latest_ios_versions.append(release_info)
+
+    return latest_ios_versions
+
+
+def update_mvt(mvt_checkout_path, latest_ios_versions):
+    version_path = os.path.join(mvt_checkout_path, "mvt/ios/data/ios_versions.json")
+    with open(version_path, "r") as version_file:
+        current_versions = json.load(version_file)
+
+    new_entry_count = 0
+    for new_version in latest_ios_versions:
+        for current_version in current_versions:
+            if new_version["build"] == current_version["build"]:
+                break
+        else:
+            # New version that does not exist in current data
+            current_versions.append(new_version)
+            new_entry_count += 1
+
+    if not new_entry_count:
+        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"])
+        )
+        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__), "../../../")
+    )
+
+    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)
+
+
+if __name__ == "__main__":
+    main()

.github/workflows/update-ios-data.yml

@@ -0,0 +1,29 @@
+name: Update iOS releases and version numbers
+run-name: ${{ github.actor }} is finding the latest iOS release version and build numbers
+on:
+  workflow_dispatch:
+  schedule:
+    # * is a special character in YAML so you have to quote this string
+    - cron:  '0 */6 * * *'
+
+
+jobs:
+  update-ios-version:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - uses: actions/setup-python@v4
+      - name: Run script to fetch latest iOS releases from Apple RSS feed.
+        run: python3 .github/workflows/scripts/update-ios-releases.py
+      - name: Create Pull Request
+        uses: peter-evans/create-pull-request@v5
+        with:
+          title: '[auto] Update iOS releases and versions'
+          commit-message: Add new iOS versions and build numbers
+          branch: auto/add-new-ios-releases
+          body: |
+            This is an automated pull request to update the iOS releases and version numbers.
+          add-paths: |
+            *.json
+          labels: |
+            automated pr

Makefile

@@ -1,5 +1,11 @@
 PWD = $(shell pwd)
 
+check:
+	flake8
+	pytest -q
+	ruff check -q .
+	black --check .
+
 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

@@ -24,7 +24,7 @@ Some recent phones will enforce the utilisation of a password to encrypt the bac
 
 ## Unpack and check the backup
 
-MVT includes a partial implementation of the Android Backup parsing, because of the implementation difference in the compression algorithm between Java and Python. The `-nocompress` option passed to adb in the section above allows to avoid this issue. You can analyse and extract SMSs containing links from the backup directly with MVT:
+MVT includes a partial implementation of the Android Backup parsing, because of the implementation difference in the compression algorithm between Java and Python. The `-nocompress` option passed to adb in the section above allows to avoid this issue. You can analyse and extract SMSs from the backup directly with MVT:
 
 ```bash
 $ mvt-android check-backup --output /path/to/results/ /path/to/backup.ab
@@ -32,7 +32,7 @@ $ mvt-android check-backup --output /path/to/results/ /path/to/backup.ab
          INFO     [mvt.android.modules.backup.sms] Running module SMS...
          INFO     [mvt.android.modules.backup.sms] Processing SMS backup file at
                   apps/com.android.providers.telephony/d_f/000000_sms_backup
-         INFO     [mvt.android.modules.backup.sms] Extracted a total of 64 SMS messages containing links
+         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.
@@ -52,4 +52,4 @@ If the backup is encrypted, ABE will prompt you to enter the password.
 
 Alternatively, [ab-decrypt](https://github.com/joernheissler/ab-decrypt) can be used for that purpose.
 
-You can then extract SMSs containing links with MVT by passing the folder path as parameter instead of the `.ab` file: `mvt-android check-backup --output /path/to/results/ /path/to/backup/` (the path to backup given should be the folder containing the `apps` folder).
+You can then extract SMSs with MVT by passing the folder path as parameter instead of the `.ab` file: `mvt-android check-backup --output /path/to/results/ /path/to/backup/` (the path to backup given should be the folder containing the `apps` folder).

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/install.md

@@ -54,7 +54,7 @@ Then you can install MVT directly from [pypi](https://pypi.org/project/mvt/)
 pip3 install mvt
 ```
 
-Or from the source code:
+If you want to have the latest features in development, you can install MVT directly from the source code. If you installed MVT previously from pypi, you should first uninstall it using `pip3 uninstall mvt` and then install from the source code:
 
 ```bash
 git clone https://github.com/mvt-project/mvt.git

docs/iocs.md

@@ -39,8 +39,10 @@ export MVT_STIX2="/home/user/IOC1.stix2:/home/user/IOC2.stix2"
 - The [Amnesty International investigations repository](https://github.com/AmnestyTech/investigations) contains STIX-formatted IOCs for:
     - [Pegasus](https://en.wikipedia.org/wiki/Pegasus_(spyware)) ([STIX2](https://raw.githubusercontent.com/AmnestyTech/investigations/master/2021-07-18_nso/pegasus.stix2))
     - [Predator from Cytrox](https://citizenlab.ca/2021/12/pegasus-vs-predator-dissidents-doubly-infected-iphone-reveals-cytrox-mercenary-spyware/) ([STIX2](https://raw.githubusercontent.com/AmnestyTech/investigations/master/2021-12-16_cytrox/cytrox.stix2))
+    - [An Android Spyware Campaign Linked to a Mercenary Company](https://github.com/AmnestyTech/investigations/tree/master/2023-03-29_android_campaign) ([STIX2](https://github.com/AmnestyTech/investigations/blob/master/2023-03-29_android_campaign/malware.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.

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/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,
+        fast_mode: 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, log=log)
+        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,
+        )
 
         self.name = "check-adb"
         self.modules = ADB_MODULES

mvt/android/cmd_check_androidqf.py

@@ -14,7 +14,6 @@ log = logging.getLogger(__name__)
 
 
 class CmdAndroidCheckAndroidQF(Command):
-
     def __init__(
         self,
         target_path: Optional[str] = None,
@@ -22,13 +21,19 @@ 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,
+        fast_mode: bool = False,
+        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,
+            fast_mode=fast_mode,
+            hashes=hashes,
+            log=log,
+        )
 
         self.name = "check-androidqf"
         self.modules = ANDROIDQF_MODULES

mvt/android/cmd_check_backup.py

@@ -9,15 +9,18 @@ import os
 import sys
 import tarfile
 from pathlib import Path
-from typing import Callable, Optional, List
+from typing import List, Optional
 
 from rich.prompt import Prompt
 
-from mvt.android.parsers.backup import (AndroidBackupParsingError,
-                                        InvalidBackupPassword, parse_ab_header,
-                                        parse_backup_file)
-from mvt.common.command import Command
 from mvt.android.modules.backup.base import BackupExtraction
+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 +28,6 @@ log = logging.getLogger(__name__)
 
 
 class CmdAndroidCheckBackup(Command):
-
     def __init__(
         self,
         target_path: Optional[str] = None,
@@ -33,13 +35,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,
+        fast_mode: bool = False,
+        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,
+            fast_mode=fast_mode,
+            hashes=hashes,
+            log=log,
+        )
 
         self.name = "check-backup"
         self.modules = BACKUP_MODULES
@@ -85,16 +93,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

@@ -6,11 +6,11 @@
 import logging
 import os
 from pathlib import Path
-from typing import Callable, Optional, List
+from typing import List, Optional
 from zipfile import ZipFile
 
-from mvt.common.command import Command
 from mvt.android.modules.bugreport.base import BugReportModule
+from mvt.common.command import Command
 
 from .modules.bugreport import BUGREPORT_MODULES
 
@@ -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,
+        fast_mode: bool = False,
+        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,
+            fast_mode=fast_mode,
+            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

@@ -21,15 +21,13 @@ log = logging.getLogger(__name__)
 class DownloadAPKs(AndroidExtraction):
     """DownloadAPKs is the main class operating the download of APKs
     from the device.
-
-
     """
 
     def __init__(
         self,
         results_path: Optional[str] = None,
-        all_apks: Optional[bool] = False,
-        packages: Optional[list] = None
+        all_apks: bool = False,
+        packages: Optional[list] = None,
     ) -> None:
         """Initialize module.
         :param results_path: Path to the folder where data should be stored
@@ -68,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
 
@@ -108,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 = []
@@ -125,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")
@@ -138,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 adb_shell.exceptions import (
+    AdbCommandFailureException,
+    DeviceAuthError,
+    UsbDeviceNotFoundError,
+    UsbReadFailedError,
+)
 from rich.prompt import Prompt
 from usb1 import USBErrorAccess, USBErrorBusy
 
-from mvt.android.parsers.backup import (InvalidBackupPassword, parse_ab_header,
-                                        parse_backup_file)
+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,
+        fast_mode: bool = False,
         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,
+            fast_mode=fast_mode,
+            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,10 @@ 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"
+        )
 
         # TODO: Base64 encoding as temporary fix to avoid byte-mangling over
         #       the shell transport...
@@ -284,19 +319,19 @@ 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.ask("Enter backup password", password=True)
             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,19 @@ class ChromeHistory(AndroidExtraction):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        fast_mode: bool = False,
         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,
+            fast_mode=fast_mode,
+            log=log,
+            results=results,
+        )
+        self.results = []
 
     def serialize(self, record: dict) -> Union[dict, list]:
         return {
@@ -38,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:
@@ -55,9 +60,11 @@ class ChromeHistory(AndroidExtraction):
         :param db_path: Path to the History database to process.
 
         """
+        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,
@@ -67,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

@@ -19,13 +19,18 @@ class DumpsysAccessibility(AndroidExtraction):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        fast_mode: bool = False,
         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,
+            fast_mode=fast_mode,
+            log=log,
+            results=results,
+        )
 
     def check_indicators(self) -> None:
         if not self.indicators:
@@ -46,8 +51,10 @@ class DumpsysAccessibility(AndroidExtraction):
         self.results = parse_dumpsys_accessibility(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

@@ -19,13 +19,18 @@ class DumpsysActivities(AndroidExtraction):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        fast_mode: bool = False,
         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,
+            fast_mode=fast_mode,
+            log=log,
+            results=results,
+        )
 
         self.results = results if results else {}
 

mvt/android/modules/adb/dumpsys_appops.py

@@ -21,13 +21,18 @@ class DumpsysAppOps(AndroidExtraction):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        fast_mode: bool = False,
         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,
+            fast_mode=fast_mode,
+            log=log,
+            results=results,
+        )
 
     def serialize(self, record: dict) -> Union[dict, list]:
         records = []
@@ -37,13 +42,15 @@ class DumpsysAppOps(AndroidExtraction):
 
             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']}",
-                    })
+                    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
 
@@ -57,10 +64,14 @@ class DumpsysAppOps(AndroidExtraction):
                     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"])
+                if (
+                    perm["name"] == "REQUEST_INSTALL_PACKAGES"
+                    and perm["access"] == "allow"
+                ):
+                    self.log.info(
+                        "Package %s with REQUEST_INSTALL_PACKAGES " "permission",
+                        result["package_name"],
+                    )
 
     def run(self) -> None:
         self._adb_connect()
@@ -69,5 +80,6 @@ class DumpsysAppOps(AndroidExtraction):
 
         self.results = parse_dumpsys_appops(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

@@ -19,13 +19,18 @@ class DumpsysBatteryDaily(AndroidExtraction):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        fast_mode: bool = False,
         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,
+            fast_mode=fast_mode,
+            log=log,
+            results=results,
+        )
 
     def serialize(self, record: dict) -> Union[dict, list]:
         return {
@@ -33,7 +38,7 @@ class DumpsysBatteryDaily(AndroidExtraction):
             "module": self.__class__.__name__,
             "event": "battery_daily",
             "data": f"Recorded update of package {record['package_name']} "
-                    f"with vers {record['vers']}"
+            f"with vers {record['vers']}",
         }
 
     def check_indicators(self) -> None:
@@ -54,5 +59,6 @@ class DumpsysBatteryDaily(AndroidExtraction):
 
         self.results = parse_dumpsys_battery_daily(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

@@ -19,13 +19,18 @@ class DumpsysBatteryHistory(AndroidExtraction):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        fast_mode: bool = False,
         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,
+            fast_mode=fast_mode,
+            log=log,
+            results=results,
+        )
 
     def check_indicators(self) -> None:
         if not self.indicators:
@@ -45,5 +50,4 @@ class DumpsysBatteryHistory(AndroidExtraction):
 
         self.results = parse_dumpsys_battery_history(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

@@ -21,13 +21,18 @@ class DumpsysDBInfo(AndroidExtraction):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        fast_mode: bool = False,
         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,
+            fast_mode=fast_mode,
+            log=log,
+            results=results,
+        )
 
     def check_indicators(self) -> None:
         if not self.indicators:
@@ -49,5 +54,7 @@ class DumpsysDBInfo(AndroidExtraction):
 
         self.results = parse_dumpsys_dbinfo(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,
+        fast_mode: bool = False,
         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,
+            fast_mode=fast_mode,
+            log=log,
+            results=results,
+        )
 
     def run(self) -> None:
         self._adb_connect()

mvt/android/modules/adb/dumpsys_receivers.py

@@ -25,13 +25,18 @@ class DumpsysReceivers(AndroidExtraction):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        fast_mode: bool = False,
         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,
+            fast_mode=fast_mode,
+            log=log,
+            results=results,
+        )
 
         self.results = results if results else {}
 
@@ -42,21 +47,31 @@ class DumpsysReceivers(AndroidExtraction):
         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"])
+                    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"])
+                    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"])
+                    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"])
+                    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"])
+                    self.log.info(
+                        'Found a receiver monitoring outgoing calls: "%s"',
+                        receiver["receiver"],
+                    )
 
                 ioc = self.indicators.check_app_id(receiver["package_name"])
                 if ioc:

mvt/android/modules/adb/files.py

@@ -30,16 +30,21 @@ class Files(AndroidExtraction):
         file_path: Optional[str] = None,
         target_path: Optional[str] = None,
         results_path: Optional[str] = None,
-        fast_mode: Optional[bool] = False,
+        fast_mode: bool = False,
         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,
+            fast_mode=fast_mode,
+            log=log,
+            results=results,
+        )
         self.full_find = False
 
-    def serialize(self, record: dict) -> Union[dict, list]:
+    def serialize(self, record: dict) -> Union[dict, list, None]:
         if "modified_time" in record:
             return {
                 "timestamp": record["modified_time"],
@@ -53,15 +58,21 @@ 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:
+        if not self.results_path:
+            return
+
         local_file_name = file_path.replace("/", "_").replace(" ", "-")
         local_files_folder = os.path.join(self.results_path, "files")
         if not os.path.exists(local_files_folder):
@@ -70,15 +81,16 @@ 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)
         if self.full_find:
             cmd = f"find '{folder}' -type f -printf '%T@ %m %s %u %g %p\n' 2> /dev/null"
             output = self._adb_command(cmd)
@@ -88,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():
@@ -119,16 +132,15 @@ 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"))
-            if self.results_path:
-                self.backup_file(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")