Bumped version

Fixed cmd_download_apks serial connection bug

.github/workflows/python-package.yml

@@ -28,7 +28,7 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        python -m pip install flake8 pytest safety stix2
+        python -m pip install flake8 pytest safety stix2 pytest-mock
         if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
         python -m pip install .
     - name: Lint with flake8

Makefile

@@ -8,3 +8,6 @@ dist:
 
 upload:
 	python3 -m twine upload dist/*
+
+test-upload:
+	python3 -m twine upload --repository testpypi dist/*

README.md

@@ -1,5 +1,5 @@
 <p align="center">
-     <img src="./docs/mvt.png" width="200" />
+     <img src="https://docs.mvt.re/en/latest/mvt.png" width="200" />
 </p>
 
 # Mobile Verification Toolkit

docs/android/download_apks.md

@@ -13,22 +13,16 @@ It might take several minutes to complete.
 !!! info
     MVT will likely warn you it was unable to download certain installed packages. There is no reason to be alarmed: this is typically expected behavior when MVT attempts to download a system package it has no privileges to access.
 
-Optionally, you can decide to enable lookups of the SHA256 hash of all the extracted APKs on [VirusTotal](https://www.virustotal.com) and/or [Koodous](https://koodous.com). While these lookups do not provide any conclusive assessment on all of the extracted APKs, they might highlight any known malicious ones:
+Optionally, you can decide to enable lookups of the SHA256 hash of all the extracted APKs on [VirusTotal](https://www.virustotal.com). While these lookups do not provide any conclusive assessment on all of the extracted APKs, they might highlight any known malicious ones:
 
 ```bash
-mvt-android download-apks --output /path/to/folder --virustotal
-mvt-android download-apks --output /path/to/folder --koodous
+MVT_VT_API_KEY=<key> mvt-android download-apks --output /path/to/folder --virustotal
 ```
 
-Or, to launch all available lookups:
+Please note that in order to use VirusTotal lookups you are required to provide your own API key through the `MVT_VT_API_KEY` environment variable. You should also note that VirusTotal enforces strict API usage. Be mindful that MVT might consume your hourly search quota.
+
+In case you have a previous extraction of APKs you want to later check against VirusTotal, you can do so with the following arguments:
 
 ```bash
-mvt-android download-apks --output /path/to/folder --all-checks
+MVT_VT_API_KEY=<key> mvt-android download-apks --from-file /path/to/folder/apks.json --virustotal
 ```
-
-In case you have a previous extraction of APKs you want to later check against VirusTotal and Koodous, you can do so with the following arguments:
-
-```bash
-mvt-android download-apks --from-file /path/to/folder/apks.json --all-checks
-```
-

docs/android/methodology.md

@@ -8,8 +8,10 @@ However, not all is lost.
 
 Because malware attacks over Android typically take the form of malicious or backdoored apps, the very first thing you might want to do is to extract and verify all installed Android packages and triage quickly if there are any which stand out as malicious or which might be atypical.
 
-While it is out of the scope of this documentation to dwell into details on how to analyze Android apps, MVT does allow to easily and automatically extract information about installed apps, download copies of them, and quickly lookup services such as [VirusTotal](https://www.virustotal.com) or [Koodous](https://koodous.com) which might quickly indicate known bad apps.
+While it is out of the scope of this documentation to dwell into details on how to analyze Android apps, MVT does allow to easily and automatically extract information about installed apps, download copies of them, and quickly look them up on services such as [VirusTotal](https://www.virustotal.com).
 
+!!! info "Using VirusTotal"
+	Please note that in order to use VirusTotal lookups you are required to provide your own API key through the `MVT_VT_API_KEY` environment variable. You should also note that VirusTotal enforces strict API usage. Be mindful that MVT might consume your hourly search quota.
 
 ## Check the device over Android Debug Bridge
 

docs/iocs.md

@@ -41,6 +41,6 @@ export MVT_STIX2="/home/user/IOC1.stix2:/home/user/IOC2.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))
 - [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).
 
-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 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.
 
 Please [open an issue](https://github.com/mvt-project/mvt/issues/) to suggest new sources of STIX-formatted IOCs.

docs/ios/records.md

@@ -18,7 +18,7 @@ If indicators are provided through the command-line, processes and domains are c
 
 ### `backup_info.json`
 
-!!! info "Availabiliy"
+!!! info "Availability"
     Backup: :material-check:  
     Full filesystem dump: :material-close:
 

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 MVT Project Developers
+copyright: Copyright © 2021-2022 MVT Project Developers
 site_description: Mobile Verification Toolkit Documentation
 markdown_extensions:
     - attr_list

mvt/android/cmd_check_adb.py

@@ -0,0 +1,25 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2022 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 mvt.common.command import Command
+
+from .modules.adb import ADB_MODULES
+
+log = logging.getLogger(__name__)
+
+
+class CmdAndroidCheckADB(Command):
+
+    name = "check-adb"
+    modules = ADB_MODULES
+
+    def __init__(self, target_path: str = None, results_path: str = None,
+                 ioc_files: list = [], module_name: str = None, serial: str = None,
+                 fast_mode: bool = False):
+        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)

mvt/android/cmd_check_backup.py

@@ -0,0 +1,85 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2022 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 io
+import logging
+import os
+import sys
+import tarfile
+from pathlib import Path
+from typing import Callable
+
+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 .modules.backup import BACKUP_MODULES
+
+log = logging.getLogger(__name__)
+
+
+class CmdAndroidCheckBackup(Command):
+
+    name = "check-backup"
+    modules = BACKUP_MODULES
+
+    def __init__(self, target_path: str = None, results_path: str = None,
+                 ioc_files: list = [], module_name: str = None, serial: str = None,
+                 fast_mode: bool = False):
+        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.backup_type = None
+        self.backup_archive = None
+        self.backup_files = []
+
+    def init(self) -> None:
+        if os.path.isfile(self.target_path):
+            self.backup_type = "ab"
+            with open(self.target_path, "rb") as handle:
+                data = handle.read()
+
+            header = parse_ab_header(data)
+            if not header["backup"]:
+                log.critical("Invalid backup format, file should be in .ab format")
+                sys.exit(1)
+
+            password = None
+            if header["encryption"] != "none":
+                password = Prompt.ask("Enter backup password", password=True)
+            try:
+                tardata = parse_backup_file(data, password=password)
+            except InvalidBackupPassword:
+                log.critical("Invalid backup password")
+                sys.exit(1)
+            except AndroidBackupParsingError as e:
+                log.critical("Impossible to parse this backup file: %s", e)
+                log.critical("Please use Android Backup Extractor (ABE) instead")
+                sys.exit(1)
+
+            dbytes = io.BytesIO(tardata)
+            self.backup_archive = tarfile.open(fileobj=dbytes)
+            for member in self.backup_archive:
+                self.backup_files.append(member.name)
+
+        elif os.path.isdir(self.target_path):
+            self.backup_type = "folder"
+            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))
+        else:
+            log.critical("Invalid backup path, path should be a folder or an Android Backup (.ab) file")
+            sys.exit(1)
+
+    def module_init(self, module: Callable) -> None:
+        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)

mvt/android/cmd_check_bugreport.py

@@ -0,0 +1,52 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2022 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
+import os
+from pathlib import Path
+from typing import Callable
+from zipfile import ZipFile
+
+from mvt.common.command import Command
+
+from .modules.bugreport import BUGREPORT_MODULES
+
+log = logging.getLogger(__name__)
+
+
+class CmdAndroidCheckBugreport(Command):
+
+    name = "check-bugreport"
+    modules = BUGREPORT_MODULES
+
+    def __init__(self, target_path: str = None, results_path: str = None,
+                 ioc_files: list = [], module_name: str = None, serial: str = None,
+                 fast_mode: bool = False):
+        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.bugreport_format = None
+        self.bugreport_archive = None
+        self.bugreport_files = []
+
+    def init(self) -> None:
+        if os.path.isfile(self.target_path):
+            self.bugreport_format = "zip"
+            self.bugreport_archive = ZipFile(self.target_path)
+            for file_name in self.bugreport_archive.namelist():
+                self.bugreport_files.append(file_name)
+        elif os.path.isdir(self.target_path):
+            self.bugreport_format = "dir"
+            parent_path = Path(self.target_path).absolute().as_posix()
+            for root, subdirs, subfiles in os.walk(os.path.abspath(self.target_path)):
+                for file_name in subfiles:
+                    self.bugreport_files.append(os.path.relpath(os.path.join(root, file_name), parent_path))
+
+    def module_init(self, module: Callable) -> None:
+        if self.bugreport_format == "zip":
+            module.from_zip(self.bugreport_archive, self.bugreport_files)
+        else:
+            module.from_folder(self.target_path, self.bugreport_files)

mvt/android/cmd_download_apks.py

@@ -6,8 +6,9 @@
 import json
 import logging
 import os
+from typing import Callable
 
-from tqdm import tqdm
+from rich.progress import track
 
 from mvt.common.module import InsufficientPrivileges
 
@@ -17,18 +18,6 @@ from .modules.adb.packages import Packages
 log = logging.getLogger(__name__)
 
 
-# TODO: Would be better to replace tqdm with rich.progress to reduce
-#       the number of dependencies. Need to investigate whether
-#       it's possible to have a similar callback system.
-class PullProgress(tqdm):
-    """PullProgress is a tqdm update system for APK downloads."""
-
-    def update_to(self, file_name, current, total):
-        if total is not None:
-            self.total = total
-        self.update(current - self.n)
-
-
 class DownloadAPKs(AndroidExtraction):
     """DownloadAPKs is the main class operating the download of APKs
     from the device.
@@ -36,22 +25,22 @@ class DownloadAPKs(AndroidExtraction):
 
     """
 
-    def __init__(self, output_folder=None, all_apks=False, log=None,
-                 packages=None):
+    def __init__(self, results_path: str = "", all_apks: bool = False,
+                 packages: list = []):
         """Initialize module.
-        :param output_folder: Path to the folder where data should be stored
+        :param results_path: Path to the folder where data should be stored
         :param all_apks: Boolean indicating whether to download all packages
                          or filter known-goods
         :param packages: Provided list of packages, typically for JSON checks
         """
-        super().__init__(output_folder=output_folder, log=log)
+        super().__init__(results_path=results_path, log=log)
 
         self.packages = packages
         self.all_apks = all_apks
-        self.output_folder_apk = None
+        self.results_path_apks = None
 
     @classmethod
-    def from_json(cls, json_path):
+    def from_json(cls, json_path: str) -> Callable:
         """Initialize this class from an existing apks.json file.
 
         :param json_path: Path to the apks.json file to parse.
@@ -61,7 +50,7 @@ class DownloadAPKs(AndroidExtraction):
             packages = json.load(handle)
             return cls(packages=packages)
 
-    def pull_package_file(self, package_name, remote_path):
+    def pull_package_file(self, package_name: str, remote_path: str) -> None:
         """Pull files related to specific package from the device.
 
         :param package_name: Name of the package to download
@@ -75,7 +64,7 @@ class DownloadAPKs(AndroidExtraction):
         if "==/" in remote_path:
             file_name = "_" + remote_path.split("==/")[1].replace(".apk", "")
 
-        local_path = os.path.join(self.output_folder_apk,
+        local_path = os.path.join(self.results_path_apks,
                                   f"{package_name}{file_name}.apk")
         name_counter = 0
         while True:
@@ -83,16 +72,13 @@ class DownloadAPKs(AndroidExtraction):
                 break
 
             name_counter += 1
-            local_path = os.path.join(self.output_folder_apk,
+            local_path = os.path.join(self.results_path_apks,
                                       f"{package_name}{file_name}_{name_counter}.apk")
 
         try:
-            with PullProgress(unit='B', unit_divisor=1024, unit_scale=True,
-                              miniters=1) as pp:
-                self._adb_download(remote_path, local_path,
-                                   progress_callback=pp.update_to)
+            self._adb_download(remote_path, local_path)
         except InsufficientPrivileges:
-            log.warn("Unable to pull package file from %s: insufficient privileges, it might be a system app",
+            log.error("Unable to pull package file from %s: insufficient privileges, it might be a system app",
                       remote_path)
             self._adb_reconnect()
             return None
@@ -104,26 +90,24 @@ class DownloadAPKs(AndroidExtraction):
 
         return local_path
 
-    def get_packages(self):
+    def get_packages(self) -> None:
         """Use the Packages adb module to retrieve the list of packages.
         We reuse the same extraction logic to then download the APKs.
-
-
         """
         self.log.info("Retrieving list of installed packages...")
 
         m = Packages()
         m.log = self.log
+        m.serial = self.serial
         m.run()
 
         self.packages = m.results
 
-    def pull_packages(self):
-        """Download all files of all selected packages from the device."""
-        log.info("Starting extraction of installed APKs at folder %s", self.output_folder)
-
-        if not os.path.exists(self.output_folder):
-            os.mkdir(self.output_folder)
+    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)
 
         # If the user provided the flag --all-apks we select all packages.
         packages_selection = []
@@ -137,7 +121,7 @@ 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",
+            log.info("Selected only %d packages which are not marked as \"system\"",
                      len(packages_selection))
 
         if len(packages_selection) == 0:
@@ -146,15 +130,15 @@ class DownloadAPKs(AndroidExtraction):
 
         log.info("Downloading packages from device. This might take some time ...")
 
-        self.output_folder_apk = os.path.join(self.output_folder, "apks")
-        if not os.path.exists(self.output_folder_apk):
-            os.mkdir(self.output_folder_apk)
+        self.results_path_apks = os.path.join(self.results_path, "apks")
+        if not os.path.exists(self.results_path_apks):
+            os.makedirs(self.results_path_apks, exist_ok=True)
 
-        counter = 0
-        for package in packages_selection:
-            counter += 1
+        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", counter, len(packages_selection),
+            log.info("[%d/%d] Package: %s", i, len(packages_selection),
                      package["package_name"])
 
             # Sometimes the package path contains multiple lines for multiple apks.
@@ -170,14 +154,12 @@ class DownloadAPKs(AndroidExtraction):
 
         log.info("Download of selected packages completed")
 
-    def save_json(self):
-        """Save the results to the package.json file."""
-        json_path = os.path.join(self.output_folder, "apks.json")
+    def save_json(self) -> None:
+        json_path = os.path.join(self.results_path, "apks.json")
         with open(json_path, "w", encoding="utf-8") as handle:
             json.dump(self.packages, handle, indent=4)
 
-    def run(self):
-        """Run all steps of fetch-apk."""
+    def run(self) -> None:
         self.get_packages()
         self._adb_connect()
         self.pull_packages()

mvt/android/lookups/__init__.py

@@ -1,4 +0,0 @@
-# Mobile Verification Toolkit (MVT)
-# Copyright (c) 2021-2022 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/lookups/koodous.py

@@ -1,58 +0,0 @@
-# Mobile Verification Toolkit (MVT)
-# Copyright (c) 2021-2022 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
-
-import requests
-from rich.console import Console
-from rich.progress import track
-from rich.table import Table
-from rich.text import Text
-
-log = logging.getLogger(__name__)
-
-
-def koodous_lookup(packages):
-    log.info("Looking up all extracted files on Koodous (www.koodous.com)")
-    log.info("This might take a while...")
-
-    table = Table(title="Koodous Packages Detections")
-    table.add_column("Package name")
-    table.add_column("File name")
-    table.add_column("Trusted")
-    table.add_column("Detected")
-    table.add_column("Rating")
-
-    total_packages = len(packages)
-    for i in track(range(total_packages), description=f"Looking up {total_packages} packages..."):
-        package = packages[i]
-        for file in package.get("files", []):
-            url = f"https://api.koodous.com/apks/{file['sha256']}"
-            res = requests.get(url)
-            report = res.json()
-
-            row = [package["package_name"], file["path"]]
-
-            if "package_name" in report:
-                trusted = "no"
-                if report["trusted"]:
-                    trusted = Text("yes", "green bold")
-
-                detected = "no"
-                if report["detected"]:
-                    detected = Text("yes", "red bold")
-
-                rating = "0"
-                if int(report["rating"]) < 0:
-                    rating = Text(str(report["rating"]), "red bold")
-
-                row.extend([trusted, detected, rating])
-            else:
-                row.extend(["n/a", "n/a", "n/a"])
-
-            table.add_row(*row)
-
-    console = Console()
-    console.print(table)

mvt/android/lookups/virustotal.py

@@ -1,100 +0,0 @@
-# Mobile Verification Toolkit (MVT)
-# Copyright (c) 2021-2022 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
-
-import requests
-from rich.console import Console
-from rich.progress import track
-from rich.table import Table
-from rich.text import Text
-
-log = logging.getLogger(__name__)
-
-
-def get_virustotal_report(hashes):
-    apikey = "233f22e200ca5822bd91103043ccac138b910db79f29af5616a9afe8b6f215ad"
-    url = f"https://www.virustotal.com/partners/sysinternals/file-reports?apikey={apikey}"
-
-    items = []
-    for sha256 in hashes:
-        items.append({
-            "autostart_location": "",
-            "autostart_entry": "",
-            "hash": sha256,
-            "local_name": "",
-            "creation_datetime": "",
-        })
-    headers = {"User-Agent": "VirusTotal", "Content-Type": "application/json"}
-    res = requests.post(url, headers=headers, json=items)
-
-    if res.status_code == 200:
-        report = res.json()
-        return report["data"]
-    else:
-        log.error("Unexpected response from VirusTotal: %s", res.status_code)
-        return None
-
-
-def virustotal_lookup(packages):
-    # NOTE: This is temporary, until we resolved the issue.
-    log.error("Unfortunately VirusTotal lookup is disabled until further notice, due to unresolved issues with the API service.")
-    return
-
-    log.info("Looking up all extracted files on VirusTotal (www.virustotal.com)")
-
-    unique_hashes = []
-    for package in packages:
-        for file in package.get("files", []):
-            if file["sha256"] not in unique_hashes:
-                unique_hashes.append(file["sha256"])
-
-    total_unique_hashes = len(unique_hashes)
-
-    detections = {}
-
-    def virustotal_query(batch):
-        report = get_virustotal_report(batch)
-        if not report:
-            return
-
-        for entry in report:
-            if entry["hash"] not in detections and entry["found"] is True:
-                detections[entry["hash"]] = entry["detection_ratio"]
-
-    batch = []
-    for i in track(range(total_unique_hashes), description=f"Looking up {total_unique_hashes} files..."):
-        file_hash = unique_hashes[i]
-        batch.append(file_hash)
-        if len(batch) == 25:
-            virustotal_query(batch)
-            batch = []
-
-    if batch:
-        virustotal_query(batch)
-
-    table = Table(title="VirusTotal Packages Detections")
-    table.add_column("Package name")
-    table.add_column("File path")
-    table.add_column("Detections")
-
-    for package in packages:
-        for file in package.get("files", []):
-            row = [package["package_name"], file["path"]]
-
-            if file["sha256"] in detections:
-                detection = detections[file["sha256"]]
-                positives = detection.split("/")[0]
-                if int(positives) > 0:
-                    row.append(Text(detection, "red bold"))
-                else:
-                    row.append(detection)
-            else:
-                row.append("not found")
-
-            table.add_row(*row)
-
-    console = Console()
-    console.print(table)

mvt/android/modules/adb/base.py

@@ -4,7 +4,6 @@
 #   https://license.mvt.re/1.1/
 
 import base64
-import getpass
 import logging
 import os
 import random
@@ -12,12 +11,14 @@ import string
 import sys
 import tempfile
 import time
+from typing import Callable
 
 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 usb1 import USBErrorAccess, USBErrorBusy
 
 from mvt.android.parsers.backup import (InvalidBackupPassword, parse_ab_header,
@@ -33,17 +34,18 @@ ADB_PUB_KEY_PATH = os.path.expanduser("~/.android/adbkey.pub")
 class AndroidExtraction(MVTModule):
     """This class provides a base for all Android extraction modules."""
 
-    def __init__(self, file_path=None, base_folder=None, output_folder=None,
-                 fast_mode=False, log=None, results=[]):
-        super().__init__(file_path=file_path, base_folder=base_folder,
-                         output_folder=output_folder, fast_mode=fast_mode,
+    def __init__(self, file_path: str = None, target_path: str = None,
+                 results_path: str = None, fast_mode: bool = False,
+                 log: logging.Logger = None, results: list = []) -> None:
+        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
 
     @staticmethod
-    def _adb_check_keys():
+    def _adb_check_keys() -> None:
         """Make sure Android adb keys exist."""
         if not os.path.isdir(os.path.dirname(ADB_KEY_PATH)):
             os.makedirs(os.path.dirname(ADB_KEY_PATH))
@@ -54,7 +56,7 @@ class AndroidExtraction(MVTModule):
         if not os.path.exists(ADB_PUB_KEY_PATH):
             write_public_keyfile(ADB_KEY_PATH, ADB_PUB_KEY_PATH)
 
-    def _adb_connect(self):
+    def _adb_connect(self) -> None:
         """Connect to the device over adb."""
         self._adb_check_keys()
 
@@ -103,17 +105,17 @@ class AndroidExtraction(MVTModule):
             else:
                 break
 
-    def _adb_disconnect(self):
+    def _adb_disconnect(self) -> None:
         """Close adb connection to the device."""
         self.device.close()
 
-    def _adb_reconnect(self):
+    def _adb_reconnect(self) -> None:
         """Reconnect to device using adb."""
         log.info("Reconnecting ...")
         self._adb_disconnect()
         self._adb_connect()
 
-    def _adb_command(self, command):
+    def _adb_command(self, command: str) -> str:
         """Execute an adb shell command.
 
         :param command: Shell command to execute
@@ -122,7 +124,7 @@ class AndroidExtraction(MVTModule):
         """
         return self.device.shell(command, read_timeout_s=200.0)
 
-    def _adb_check_if_root(self):
+    def _adb_check_if_root(self) -> bool:
         """Check if we have a `su` binary on the Android device.
 
 
@@ -131,7 +133,7 @@ class AndroidExtraction(MVTModule):
         """
         return bool(self._adb_command("command -v su"))
 
-    def _adb_root_or_die(self):
+    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!")
@@ -145,7 +147,7 @@ class AndroidExtraction(MVTModule):
         """
         return self._adb_command(f"su -c {command}")
 
-    def _adb_check_file_exists(self, file):
+    def _adb_check_file_exists(self, file: str) -> bool:
         """Verify that a file exists.
 
         :param file: Path of the file
@@ -162,7 +164,9 @@ class AndroidExtraction(MVTModule):
 
         return bool(self._adb_command_as_root(f"[ ! -f {file} ] || echo 1"))
 
-    def _adb_download(self, remote_path, local_path, progress_callback=None, retry_root=True):
+    def _adb_download(self, remote_path: str, local_path: str,
+                      progress_callback: Callable = None,
+                      retry_root: bool = True) -> None:
         """Download a file form the device.
 
         :param remote_path: Path to download from the device
@@ -179,7 +183,8 @@ class AndroidExtraction(MVTModule):
             else:
                 raise Exception(f"Unable to download file {remote_path}: {e}")
 
-    def _adb_download_root(self, remote_path, local_path, progress_callback=None):
+    def _adb_download_root(self, remote_path: str, local_path: str,
+                           progress_callback: Callable = None) -> None:
         try:
             # Check if we have root, if not raise an Exception.
             self._adb_root_or_die()
@@ -207,7 +212,8 @@ class AndroidExtraction(MVTModule):
         except AdbCommandFailureException as e:
             raise Exception(f"Unable to download file {remote_path}: {e}")
 
-    def _adb_process_file(self, remote_path, process_routine):
+    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.
 
@@ -247,7 +253,7 @@ class AndroidExtraction(MVTModule):
         # Disconnect from the device.
         self._adb_disconnect()
 
-    def _generate_backup(self, package_name):
+    def _generate_backup(self, package_name: str) -> bytes:
         self.log.warning("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...
@@ -264,7 +270,7 @@ class AndroidExtraction(MVTModule):
             return parse_backup_file(backup_output, password=None)
 
         for password_retry in range(0, 3):
-            backup_password = getpass.getpass(prompt="Backup password: ", stream=None)
+            backup_password = Prompt.ask("Enter backup password", password=True)
             try:
                 decrypted_backup_tar = parse_backup_file(backup_output, backup_password)
                 return decrypted_backup_tar
@@ -273,6 +279,6 @@ class AndroidExtraction(MVTModule):
 
         self.log.warn("All attempts to decrypt backup with password failed!")
 
-    def run(self):
+    def run(self) -> None:
         """Run the main procedure."""
         raise NotImplementedError

mvt/android/modules/adb/chrome_history.py

@@ -20,13 +20,14 @@ CHROME_HISTORY_PATH = "data/data/com.android.chrome/app_chrome/Default/History"
 class ChromeHistory(AndroidExtraction):
     """This module extracts records from Android's Chrome browsing history."""
 
-    def __init__(self, file_path=None, base_folder=None, output_folder=None,
-                 serial=None, fast_mode=False, log=None, results=[]):
-        super().__init__(file_path=file_path, base_folder=base_folder,
-                         output_folder=output_folder, fast_mode=fast_mode,
+    def __init__(self, file_path: str = None, target_path: str = None,
+                 results_path: str = None, fast_mode: bool = False,
+                 log: logging.Logger = None, results: list = []) -> 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):
+    def serialize(self, record: dict) -> None:
         return {
             "timestamp": record["isodate"],
             "module": self.__class__.__name__,
@@ -34,7 +35,7 @@ class ChromeHistory(AndroidExtraction):
             "data": f"{record['id']} - {record['url']} (visit ID: {record['visit_id']}, redirect source: {record['redirect_source']})"
         }
 
-    def check_indicators(self):
+    def check_indicators(self) -> None:
         if not self.indicators:
             return
 
@@ -42,7 +43,7 @@ class ChromeHistory(AndroidExtraction):
             if self.indicators.check_domain(result["url"]):
                 self.detected.append(result)
 
-    def _parse_db(self, db_path):
+    def _parse_db(self, db_path: str) -> None:
         """Parse a Chrome History database file.
 
         :param db_path: Path to the History database to process.
@@ -77,7 +78,7 @@ class ChromeHistory(AndroidExtraction):
 
         log.info("Extracted a total of %d history items", len(self.results))
 
-    def run(self):
+    def run(self) -> None:
         try:
             self._adb_process_file(os.path.join("/", CHROME_HISTORY_PATH),
                                    self._parse_db)

mvt/android/modules/adb/dumpsys_accessibility.py

@@ -15,13 +15,14 @@ log = logging.getLogger(__name__)
 class DumpsysAccessibility(AndroidExtraction):
     """This module extracts stats on accessibility."""
 
-    def __init__(self, file_path=None, base_folder=None, output_folder=None,
-                 serial=None, fast_mode=False, log=None, results=[]):
-        super().__init__(file_path=file_path, base_folder=base_folder,
-                         output_folder=output_folder, fast_mode=fast_mode,
+    def __init__(self, file_path: str = None, target_path: str = None,
+                 results_path: str = None, fast_mode: bool = False,
+                 log: logging.Logger = None, results: list = []) -> 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):
+    def check_indicators(self) -> None:
         if not self.indicators:
             return
 
@@ -32,7 +33,7 @@ class DumpsysAccessibility(AndroidExtraction):
                 self.detected.append(result)
                 continue
 
-    def run(self):
+    def run(self) -> None:
         self._adb_connect()
         output = self._adb_command("dumpsys accessibility")
         self._adb_disconnect()

mvt/android/modules/adb/dumpsys_activities.py

@@ -15,15 +15,16 @@ log = logging.getLogger(__name__)
 class DumpsysActivities(AndroidExtraction):
     """This module extracts details on receivers for risky activities."""
 
-    def __init__(self, file_path=None, base_folder=None, output_folder=None,
-                 serial=None, fast_mode=False, log=None, results=[]):
-        super().__init__(file_path=file_path, base_folder=base_folder,
-                         output_folder=output_folder, fast_mode=fast_mode,
+    def __init__(self, file_path: str = None, target_path: str = None,
+                 results_path: str = None, fast_mode: bool = False,
+                 log: logging.Logger = None, results: list = []) -> None:
+        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 {}
 
-    def check_indicators(self):
+    def check_indicators(self) -> None:
         if not self.indicators:
             return
 
@@ -35,7 +36,7 @@ class DumpsysActivities(AndroidExtraction):
                     self.detected.append({intent: activity})
                     continue
 
-    def run(self):
+    def run(self) -> None:
         self._adb_connect()
         output = self._adb_command("dumpsys package")
         self._adb_disconnect()

mvt/android/modules/adb/dumpsys_appops.py

@@ -4,7 +4,6 @@
 #   https://license.mvt.re/1.1/
 
 import logging
-import re
 
 from mvt.android.parsers.dumpsys import parse_dumpsys_appops
 
@@ -18,13 +17,14 @@ class DumpsysAppOps(AndroidExtraction):
 
     slug = "dumpsys_appops"
 
-    def __init__(self, file_path=None, base_folder=None, output_folder=None,
-                 serial=None, fast_mode=False, log=None, results=[]):
-        super().__init__(file_path=file_path, base_folder=base_folder,
-                         output_folder=output_folder, fast_mode=fast_mode,
+    def __init__(self, file_path: str = None, target_path: str = None,
+                 results_path: str = None, fast_mode: bool = False,
+                 log: logging.Logger = None, results: list = []) -> 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):
+    def serialize(self, record: dict) -> None:
         records = []
         for perm in record["permissions"]:
             if "entries" not in perm:
@@ -36,12 +36,12 @@ class DumpsysAppOps(AndroidExtraction):
                         "timestamp": entry["timestamp"],
                         "module": self.__class__.__name__,
                         "event": entry["access"],
-                        "data": f"{record['package_name']} access to {perm['name']} : {entry['access']}",
+                        "data": f"{record['package_name']} access to {perm['name']}: {entry['access']}",
                     })
 
         return records
 
-    def check_indicators(self):
+    def check_indicators(self) -> None:
         for result in self.results:
             if self.indicators:
                 ioc = self.indicators.check_app_id(result.get("package_name"))
@@ -55,7 +55,7 @@ class DumpsysAppOps(AndroidExtraction):
                     self.log.info("Package %s with REQUEST_INSTALL_PACKAGES permission",
                                   result["package_name"])
 
-    def run(self):
+    def run(self) -> None:
         self._adb_connect()
         output = self._adb_command("dumpsys appops")
         self._adb_disconnect()

mvt/android/modules/adb/dumpsys_battery_daily.py

@@ -15,13 +15,14 @@ log = logging.getLogger(__name__)
 class DumpsysBatteryDaily(AndroidExtraction):
     """This module extracts records from battery daily updates."""
 
-    def __init__(self, file_path=None, base_folder=None, output_folder=None,
-                 serial=None, fast_mode=False, log=None, results=[]):
-        super().__init__(file_path=file_path, base_folder=base_folder,
-                         output_folder=output_folder, fast_mode=fast_mode,
+    def __init__(self, file_path: str = None, target_path: str = None,
+                 results_path: str = None, fast_mode: bool = False,
+                 log: logging.Logger = None, results: list = []) -> 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):
+    def serialize(self, record: dict) -> None:
         return {
             "timestamp": record["from"],
             "module": self.__class__.__name__,
@@ -29,7 +30,7 @@ class DumpsysBatteryDaily(AndroidExtraction):
             "data": f"Recorded update of package {record['package_name']} with vers {record['vers']}"
         }
 
-    def check_indicators(self):
+    def check_indicators(self) -> None:
         if not self.indicators:
             return
 
@@ -40,7 +41,7 @@ class DumpsysBatteryDaily(AndroidExtraction):
                 self.detected.append(result)
                 continue
 
-    def run(self):
+    def run(self) -> None:
         self._adb_connect()
         output = self._adb_command("dumpsys batterystats --daily")
         self._adb_disconnect()

mvt/android/modules/adb/dumpsys_battery_history.py

@@ -15,13 +15,14 @@ log = logging.getLogger(__name__)
 class DumpsysBatteryHistory(AndroidExtraction):
     """This module extracts records from battery history events."""
 
-    def __init__(self, file_path=None, base_folder=None, output_folder=None,
-                 serial=None, fast_mode=False, log=None, results=[]):
-        super().__init__(file_path=file_path, base_folder=base_folder,
-                         output_folder=output_folder, fast_mode=fast_mode,
+    def __init__(self, file_path: str = None, target_path: str = None,
+                 results_path: str = None, fast_mode: bool = False,
+                 log: logging.Logger = None, results: list = []) -> 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):
+    def check_indicators(self) -> None:
         if not self.indicators:
             return
 
@@ -32,7 +33,7 @@ class DumpsysBatteryHistory(AndroidExtraction):
                 self.detected.append(result)
                 continue
 
-    def run(self):
+    def run(self) -> None:
         self._adb_connect()
         output = self._adb_command("dumpsys batterystats --history")
         self._adb_disconnect()

mvt/android/modules/adb/dumpsys_dbinfo.py

@@ -4,7 +4,6 @@
 #   https://license.mvt.re/1.1/
 
 import logging
-import re
 
 from mvt.android.parsers import parse_dumpsys_dbinfo
 
@@ -18,13 +17,14 @@ class DumpsysDBInfo(AndroidExtraction):
 
     slug = "dumpsys_dbinfo"
 
-    def __init__(self, file_path=None, base_folder=None, output_folder=None,
-                 serial=None, fast_mode=False, log=None, results=[]):
-        super().__init__(file_path=file_path, base_folder=base_folder,
-                         output_folder=output_folder, fast_mode=fast_mode,
+    def __init__(self, file_path: str = None, target_path: str = None,
+                 results_path: str = None, fast_mode: bool = False,
+                 log: logging.Logger = None, results: list = []) -> 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):
+    def check_indicators(self) -> None:
         if not self.indicators:
             return
 
@@ -37,7 +37,7 @@ class DumpsysDBInfo(AndroidExtraction):
                     self.detected.append(result)
                     continue
 
-    def run(self):
+    def run(self) -> None:
         self._adb_connect()
         output = self._adb_command("dumpsys dbinfo")
         self._adb_disconnect()

mvt/android/modules/adb/dumpsys_full.py

@@ -14,18 +14,19 @@ log = logging.getLogger(__name__)
 class DumpsysFull(AndroidExtraction):
     """This module extracts stats on battery consumption by processes."""
 
-    def __init__(self, file_path=None, base_folder=None, output_folder=None,
-                 serial=None, fast_mode=False, log=None, results=[]):
-        super().__init__(file_path=file_path, base_folder=base_folder,
-                         output_folder=output_folder, fast_mode=fast_mode,
+    def __init__(self, file_path: str = None, target_path: str = None,
+                 results_path: str = None, fast_mode: bool = False,
+                 log: logging.Logger = None, results: list = []) -> None:
+        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):
+    def run(self) -> None:
         self._adb_connect()
 
         output = self._adb_command("dumpsys")
-        if self.output_folder:
-            output_path = os.path.join(self.output_folder, "dumpsys.txt")
+        if self.results_path:
+            output_path = os.path.join(self.results_path, "dumpsys.txt")
             with open(output_path, "w", encoding="utf-8") as handle:
                 handle.write(output)
 

mvt/android/modules/adb/dumpsys_receivers.py

@@ -21,15 +21,16 @@ INTENT_NEW_OUTGOING_CALL = "android.intent.action.NEW_OUTGOING_CALL"
 class DumpsysReceivers(AndroidExtraction):
     """This module extracts details on receivers for risky activities."""
 
-    def __init__(self, file_path=None, base_folder=None, output_folder=None,
-                 serial=None, fast_mode=False, log=None, results=[]):
-        super().__init__(file_path=file_path, base_folder=base_folder,
-                         output_folder=output_folder, fast_mode=fast_mode,
+    def __init__(self, file_path: str = None, target_path: str = None,
+                 results_path: str = None, fast_mode: bool = False,
+                 log: logging.Logger = None, results: list = []) -> None:
+        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 {}
 
-    def check_indicators(self):
+    def check_indicators(self) -> None:
         if not self.indicators:
             return
 
@@ -57,7 +58,7 @@ class DumpsysReceivers(AndroidExtraction):
                 self.detected.append({intent: receiver})
                 continue
 
-    def run(self):
+    def run(self) -> None:
         self._adb_connect()
 
         output = self._adb_command("dumpsys package")

mvt/android/modules/adb/files.py

@@ -5,6 +5,7 @@
 
 import datetime
 import logging
+import os
 import stat
 
 from mvt.common.utils import convert_timestamp_to_iso
@@ -13,24 +14,71 @@ from .base import AndroidExtraction
 
 log = logging.getLogger(__name__)
 
+ANDROID_TMP_FOLDERS = [
+    "/tmp/",
+    "/data/local/tmp/",
+]
+ANDROID_MEDIA_FOLDERS = [
+    "/data/media/0",
+    "/sdcard/",
+]
+
 
 class Files(AndroidExtraction):
     """This module extracts the list of files on the device."""
 
-    def __init__(self, file_path=None, base_folder=None, output_folder=None,
-                 serial=None, fast_mode=False, log=None, results=[]):
-        super().__init__(file_path=file_path, base_folder=base_folder,
-                         output_folder=output_folder, fast_mode=fast_mode,
+    def __init__(self, file_path: str = None, target_path: str = None,
+                 results_path: str = None, fast_mode: bool = False,
+                 log: logging.Logger = None, results: list = []) -> None:
+        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 find_files(self, folder):
+    def serialize(self, record: dict) -> None:
+        if "modified_time" in record:
+            return {
+                "timestamp": record["modified_time"],
+                "module": self.__class__.__name__,
+                "event": "file_modified",
+                "data": record["path"],
+            }
+
+    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"])
+
+            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.detected.append(result)
+
+    def backup_file(self, file_path: str) -> None:
+        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):
+            os.mkdir(local_files_folder)
+
+        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)
+        except Exception:
+            pass
+        else:
+            self.log.info("Downloaded file %s to local copy at %s",
+                          file_path, local_file_path)
+
+    def find_files(self, folder: str) -> None:
         if self.full_find:
-            output = self._adb_command(f"find '{folder}' -printf '%T@ %m %s %u %g %p\n' 2> /dev/null")
+            output = self._adb_command(f"find '{folder}' -type f -printf '%T@ %m %s %u %g %p\n' 2> /dev/null")
 
             for file_line in output.splitlines():
                 [unix_timestamp, mode, size, owner, group, full_path] = file_line.rstrip().split(" ", 5)
                 mod_time = convert_timestamp_to_iso(datetime.datetime.utcfromtimestamp(int(float(unix_timestamp))))
+
                 self.results.append({
                     "path": full_path,
                     "modified_time": mod_time,
@@ -42,56 +90,38 @@ class Files(AndroidExtraction):
                     "group": group,
                 })
         else:
-            output = self._adb_command(f"find '{folder}' 2> /dev/null")
+            output = self._adb_command(f"find '{folder}' -type f 2> /dev/null")
             for file_line in output.splitlines():
                 self.results.append({"path": file_line.rstrip()})
 
-    def serialize(self, record):
-        if "modified_time" in record:
-            return {
-                "timestamp": record["modified_time"],
-                "module": self.__class__.__name__,
-                "event": "file_modified",
-                "data": record["path"],
-            }
-
-    def check_suspicious(self):
-        """Check for files with suspicious permissions"""
-        for result in sorted(self.results, key=lambda item: item["path"]):
-            if result.get("is_suid"):
-                self.log.warning("Found an SUID file in a non-standard directory \"%s\".",
-                                 result["path"])
-                self.detected.append(result)
-
-    def check_indicators(self):
-        """Check file list for known suspicious files or suspicious properties"""
-        self.check_suspicious()
-
-        if not self.indicators:
-            return
-
-        for result in self.results:
-            if self.indicators.check_file_path(result["path"]):
-                self.log.warning("Found a known suspicous file at path: \"%s\"", result["path"])
-                self.detected.append(result)
-
-    def run(self):
+    def run(self) -> None:
         self._adb_connect()
 
         output = self._adb_command("find '/' -maxdepth 1 -printf '%T@ %m %s %u %g %p\n' 2> /dev/null")
         if output or output.strip().splitlines():
             self.full_find = True
 
-        for data_path in ["/data/local/tmp/", "/sdcard/", "/tmp/"]:
-            self.find_files(data_path)
+        for tmp_folder in ANDROID_TMP_FOLDERS:
+            self.find_files(tmp_folder)
 
-        self.log.info("Found %s files in primary Android data directories", len(self.results))
+        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"))
+
+        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))
 
         if self.fast_mode:
             self.log.info("Flag --fast was enabled: skipping full file listing")
         else:
             self.log.info("Processing full file listing. This may take a while...")
             self.find_files("/")
+
             self.log.info("Found %s total files", len(self.results))
 
         self._adb_disconnect()

mvt/android/modules/adb/getprop.py

@@ -4,7 +4,6 @@
 #   https://license.mvt.re/1.1/
 
 import logging
-import re
 from datetime import datetime, timedelta
 
 from mvt.android.parsers import parse_getprop
@@ -17,15 +16,16 @@ log = logging.getLogger(__name__)
 class Getprop(AndroidExtraction):
     """This module extracts device properties from getprop command."""
 
-    def __init__(self, file_path=None, base_folder=None, output_folder=None,
-                 serial=None, fast_mode=False, log=None, results=[]):
-        super().__init__(file_path=file_path, base_folder=base_folder,
-                         output_folder=output_folder, fast_mode=fast_mode,
+    def __init__(self, file_path: str = None, target_path: str = None,
+                 results_path: str = None, fast_mode: bool = False,
+                 log: logging.Logger = None, results: list = []) -> None:
+        super().__init__(file_path=file_path, target_path=target_path,
+                         results_path=results_path, fast_mode=fast_mode,
                          log=log, results=results)
 
         self.results = {} if not results else results
 
-    def run(self):
+    def run(self) -> None:
         self._adb_connect()
         output = self._adb_command("getprop")
         self._adb_disconnect()

mvt/android/modules/adb/logcat.py

@@ -14,13 +14,14 @@ log = logging.getLogger(__name__)
 class Logcat(AndroidExtraction):
     """This module extracts details on installed packages."""
 
-    def __init__(self, file_path=None, base_folder=None, output_folder=None,
-                 serial=None, fast_mode=False, log=None, results=[]):
-        super().__init__(file_path=file_path, base_folder=base_folder,
-                         output_folder=output_folder, fast_mode=fast_mode,
+    def __init__(self, file_path: str = None, target_path: str = None,
+                 results_path: str = None, fast_mode: bool = False,
+                 log: logging.Logger = None, results: list = []) -> None:
+        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):
+    def run(self) -> None:
         self._adb_connect()
 
         # Get the current logcat.
@@ -28,8 +29,8 @@ class Logcat(AndroidExtraction):
         # Get the locat prior to last reboot.
         last_output = self._adb_command("logcat -L")
 
-        if self.output_folder:
-            logcat_path = os.path.join(self.output_folder,
+        if self.results_path:
+            logcat_path = os.path.join(self.results_path,
                                        "logcat.txt")
             with open(logcat_path, "w", encoding="utf-8") as handle:
                 handle.write(output)
@@ -37,7 +38,7 @@ class Logcat(AndroidExtraction):
             log.info("Current logcat logs stored at %s",
                      logcat_path)
 
-            logcat_last_path = os.path.join(self.output_folder,
+            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)

mvt/android/modules/adb/packages.py

@@ -4,12 +4,13 @@
 #   https://license.mvt.re/1.1/
 
 import logging
-import os
 
-import pkg_resources
+from rich.console import Console
+from rich.progress import track
+from rich.table import Table
+from rich.text import Text
 
-from mvt.android.lookups.koodous import koodous_lookup
-from mvt.android.lookups.virustotal import virustotal_lookup
+from mvt.common.virustotal import VTNoKey, VTQuotaExceeded, virustotal_lookup
 
 from .base import AndroidExtraction
 
@@ -71,13 +72,14 @@ ROOT_PACKAGES = [
 class Packages(AndroidExtraction):
     """This module extracts the list of installed packages."""
 
-    def __init__(self, file_path=None, base_folder=None, output_folder=None,
-                 serial=None, fast_mode=False, log=None, results=[]):
-        super().__init__(file_path=file_path, base_folder=base_folder,
-                         output_folder=output_folder, fast_mode=fast_mode,
+    def __init__(self, file_path: str = None, target_path: str = None,
+                 results_path: str = None, fast_mode: bool = False,
+                 log: logging.Logger = None, results: list = []) -> 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):
+    def serialize(self, record: dict) -> None:
         records = []
 
         timestamps = [
@@ -96,7 +98,7 @@ class Packages(AndroidExtraction):
 
         return records
 
-    def check_indicators(self):
+    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\"",
@@ -113,14 +115,67 @@ class Packages(AndroidExtraction):
                 self.detected.append(result)
                 continue
 
-            for package_file in result["files"]:
+            for package_file in result.get("files", []):
                 ioc = self.indicators.check_file_hash(package_file["sha256"])
                 if ioc:
                     result["matched_indicator"] = ioc
                     self.detected.append(result)
 
     @staticmethod
-    def parse_package_for_details(output):
+    def check_virustotal(packages: list) -> None:
+        hashes = []
+        for package in packages:
+            for file in package.get("files", []):
+                if file["sha256"] not in hashes:
+                    hashes.append(file["sha256"])
+
+        total_hashes = len(hashes)
+        detections = {}
+
+        for i in track(range(total_hashes), description=f"Looking up {total_hashes} files..."):
+            try:
+                results = virustotal_lookup(hashes[i])
+            except VTNoKey as e:
+                log.info(e)
+                return
+            except VTQuotaExceeded as e:
+                log.error("Unable to continue: %s", e)
+                break
+
+            if not results:
+                continue
+
+            positives = results["attributes"]["last_analysis_stats"]["malicious"]
+            total = len(results["attributes"]["last_analysis_results"])
+
+            detections[hashes[i]] = f"{positives}/{total}"
+
+        table = Table(title="VirusTotal Packages Detections")
+        table.add_column("Package name")
+        table.add_column("File path")
+        table.add_column("Detections")
+
+        for package in packages:
+            for file in package.get("files", []):
+                row = [package["package_name"], file["path"]]
+
+                if file["sha256"] in detections:
+                    detection = detections[file["sha256"]]
+                    positives = detection.split("/")[0]
+                    if int(positives) > 0:
+                        row.append(Text(detection, "red bold"))
+                    else:
+                        row.append(detection)
+                else:
+                    row.append("not found")
+
+                table.add_row(*row)
+
+        console = Console()
+        console.print(table)
+
+    @staticmethod
+    def parse_package_for_details(output: str) -> dict:
         details = {
             "uid": "",
             "version_name": "",
@@ -159,7 +214,7 @@ class Packages(AndroidExtraction):
 
         return details
 
-    def _get_files_for_package(self, package_name):
+    def _get_files_for_package(self, package_name: str) -> list:
         output = self._adb_command(f"pm path {package_name}")
         output = output.strip().replace("package:", "")
         if not output:
@@ -184,7 +239,7 @@ class Packages(AndroidExtraction):
 
         return package_files
 
-    def run(self):
+    def run(self) -> None:
         self._adb_connect()
 
         packages = self._adb_command("pm list packages -u -i -f")
@@ -262,8 +317,7 @@ class Packages(AndroidExtraction):
                           result["package_name"], result["installer"], result["timestamp"])
 
         if not self.fast_mode:
-            virustotal_lookup(packages_to_lookup)
-            koodous_lookup(packages_to_lookup)
+            self.check_virustotal(packages_to_lookup)
 
         self.log.info("Extracted at total of %d installed package names",
                       len(self.results))

mvt/android/modules/adb/processes.py

@@ -13,13 +13,14 @@ log = logging.getLogger(__name__)
 class Processes(AndroidExtraction):
     """This module extracts details on running processes."""
 
-    def __init__(self, file_path=None, base_folder=None, output_folder=None,
-                 serial=None, fast_mode=False, log=None, results=[]):
-        super().__init__(file_path=file_path, base_folder=base_folder,
-                         output_folder=output_folder, fast_mode=fast_mode,
+    def __init__(self, file_path: str = None, target_path: str = None,
+                 results_path: str = None, fast_mode: bool = False,
+                 log: logging.Logger = None, results: list = []) -> 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):
+    def check_indicators(self) -> None:
         if not self.indicators:
             return
 
@@ -29,7 +30,7 @@ class Processes(AndroidExtraction):
                 result["matched_indicator"] = ioc
                 self.detected.append(result)
 
-    def run(self):
+    def run(self) -> None:
         self._adb_connect()
 
         output = self._adb_command("ps -e")

mvt/android/modules/adb/root_binaries.py

@@ -13,13 +13,14 @@ log = logging.getLogger(__name__)
 class RootBinaries(AndroidExtraction):
     """This module extracts the list of installed packages."""
 
-    def __init__(self, file_path=None, base_folder=None, output_folder=None,
-                 serial=None, fast_mode=False, log=None, results=[]):
-        super().__init__(file_path=file_path, base_folder=base_folder,
-                         output_folder=output_folder, fast_mode=fast_mode,
+    def __init__(self, file_path: str = None, target_path: str = None,
+                 results_path: str = None, fast_mode: bool = False,
+                 log: logging.Logger = None, results: list = []) -> None:
+        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):
+    def run(self) -> None:
         root_binaries = [
             "su",
             "busybox",