Merge branch 'main' into add-shell-completion-command

Bumps [pydantic-settings](https://github.com/pydantic/pydantic-settings) from 2.13.1 to 2.14.0.
- [Release notes](https://github.com/pydantic/pydantic-settings/releases)
- [Commits](https://github.com/pydantic/pydantic-settings/compare/v2.13.1...v2.14.0)

---
updated-dependencies:
- dependency-name: pydantic-settings
  dependency-version: 2.14.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

.github/workflows/mypy.yml

@@ -1,5 +1,8 @@
 name: Mypy
-on: workflow_dispatch
+on:
+  pull_request:
+    branches: [main]
+  workflow_dispatch:
 
 jobs:
   mypy_py3:

README.md

@@ -60,6 +60,28 @@ For alternative installation options and known issues, please refer to the [docu
 
 MVT provides two commands `mvt-ios` and `mvt-android`. [Check out the documentation to learn how to use them!](https://docs.mvt.re/)
 
+### Shell completion
+
+MVT can generate shell completion scripts for Bash, Zsh, and Fish:
+
+```bash
+mvt-ios completion
+mvt-android completion
+```
+
+The commands print setup instructions by default. To generate a completion script directly, pass the shell name:
+
+```bash
+mvt-ios completion bash
+mvt-android completion zsh
+```
+
+MVT only writes completion files or shell configuration when `--install` is passed. See the [command completion documentation](https://docs.mvt.re/en/latest/command_completion/) for details.
+Module-running `check-*` commands can load custom Python modules with
+`--load-module PATH` or from a folder set in `MVT_CUSTOM_MODULES`. See the
+[development documentation](https://docs.mvt.re/en/latest/development/) for
+details.
+
 
 ## License
 

docs/command_completion.md

@@ -1,43 +1,66 @@
-# Command Completion 
+# Command Completion
 
-MVT utilizes the [Click](https://click.palletsprojects.com/en/stable/) library for creating its command line interface. 
+MVT utilizes the [Click](https://click.palletsprojects.com/en/stable/) library for creating its command line interface.
 
 Click provides tab completion support for Bash (version 4.4 and up), Zsh, and Fish.
 
-To enable it, you need to manually register a special function with your shell, which varies depending on the shell you are using.
+To enable it, you need to register a completion script with your shell, which varies depending on the shell you are using.
 
-The following describes how to generate the command completion scripts and add them to your shell configuration. 
+The following describes how to generate the command completion scripts and add them to your shell configuration.
 
 > **Note: You will need to start a new shell for the changes to take effect.**
 
 ### For Bash
 
 ```bash
-# Generates bash completion scripts
-echo "$(_MVT_IOS_COMPLETE=bash_source mvt-ios)" > ~/.mvt-ios-complete.bash &&
-echo "$(_MVT_ANDROID_COMPLETE=bash_source mvt-android)" > ~/.mvt-android-complete.bash
+# Generate bash completion scripts
+mvt-ios completion bash > ~/.mvt-ios-complete.bash
+mvt-android completion bash > ~/.mvt-android-complete.bash
 ```
 
 Add the following to `~/.bashrc`:
 ```bash
 # source mvt completion scripts
-. ~/.mvt-ios-complete.bash && .  ~/.mvt-android-complete.bash
+[ -f ~/.mvt-ios-complete.bash ] && . ~/.mvt-ios-complete.bash
+[ -f ~/.mvt-android-complete.bash ] && . ~/.mvt-android-complete.bash
 ```
 
 ### For Zsh
 
 ```bash
-# Generates zsh completion scripts
-echo "$(_MVT_IOS_COMPLETE=zsh_source mvt-ios)" >  ~/.mvt-ios-complete.zsh &&
-echo "$(_MVT_ANDROID_COMPLETE=zsh_source mvt-android)" > ~/.mvt-android-complete.zsh
+# Generate zsh completion scripts
+mvt-ios completion zsh > ~/.mvt-ios-complete.zsh
+mvt-android completion zsh > ~/.mvt-android-complete.zsh
 ```
 
 Add the following to `~/.zshrc`:
 ```bash
 # source mvt completion scripts
-.  ~/.mvt-ios-complete.zsh  && .  ~/.mvt-android-complete.zsh
+[ -f ~/.mvt-ios-complete.zsh ] && . ~/.mvt-ios-complete.zsh
+[ -f ~/.mvt-android-complete.zsh ] && . ~/.mvt-android-complete.zsh
 ```
 
+### For Fish
+
+```bash
+# Generate fish completion scripts
+mkdir -p ~/.config/fish/completions
+mvt-ios completion fish > ~/.config/fish/completions/mvt-ios.fish
+mvt-android completion fish > ~/.config/fish/completions/mvt-android.fish
+```
+
+Fish loads completion files from `~/.config/fish/completions` automatically.
+
+### Automatic Installation
+
+MVT can write the completion file and update the relevant shell configuration for Bash and Zsh when you pass `--install`:
+
+```bash
+mvt-ios completion bash --install
+mvt-android completion bash --install
+```
+
+Replace `bash` with `zsh` or `fish` as needed. For Fish, `--install` writes the completion file into `~/.config/fish/completions`.
+
 For more information, visit the official [Click Docs](https://click.palletsprojects.com/en/stable/shell-completion/#enabling-completion).
 
-

docs/development.md

@@ -21,6 +21,100 @@ make check
 
 Run these tests before making new commits or opening pull requests.
 
+## Module dependencies
+
+Modules can require other modules to run first by declaring their classes in
+`dependencies`. The command runner uses a stable topological ordering, so the
+existing module list order is preserved wherever dependency constraints allow.
+
+```python
+class DependentModule(MVTModule):
+    dependencies = (PrerequisiteModule,)
+
+    def run(self):
+        prerequisite_results = self.get_dependency_results(PrerequisiteModule)
+```
+
+Selecting a single module also runs its transitive dependencies. If a dependency
+is unavailable or the dependency graph contains a cycle, the command logs a
+warning and does not run any modules.
+
+## Custom modules
+
+Module-running `check-*` commands can load custom modules from Python files that
+are not installed as part of MVT. Load one file with:
+
+```bash
+mvt-ios check-backup --load-module ./example_module.py --output ./out ./backup
+```
+
+You can also load a folder. MVT loads non-hidden top-level `*.py` files in
+sorted order and skips `__init__.py`:
+
+```bash
+mvt-ios check-fs --load-module ./custom_modules ./filesystem-dump
+```
+
+Set `MVT_CUSTOM_MODULES` to load a folder for every module-running command. This
+folder is loaded before any `--load-module` path:
+
+```bash
+MVT_CUSTOM_MODULES=./custom_modules mvt-android check-bugreport ./bugreport.zip
+```
+
+Custom modules are normal `MVTModule` subclasses:
+
+```python
+from mvt.common.module import MVTModule
+
+
+class ExampleCustomModule(MVTModule):
+    supported_commands = (("ios", "check-backup"), ("ios", "check-fs"))
+    slug = "example_custom_module"
+
+    def run(self):
+        self.results = [{"message": "custom module ran"}]
+
+    def check_indicators(self):
+        pass
+
+    def serialize(self, result):
+        return None
+```
+
+Use `supported_commands` to restrict a module to specific platform/command
+pairs. Missing or empty `supported_commands` means the module is available to
+all commands, which keeps older modules compatible. Supported pairs are:
+
+```python
+("ios", "check-backup")
+("ios", "check-fs")
+("ios", "check-iocs")
+("android", "check-backup")
+("android", "check-bugreport")
+("android", "check-androidqf")
+("android", "check-intrusion-logs")
+("android", "check-iocs")
+```
+
+Custom modules can depend on existing MVT module classes. Dependencies are
+resolved with the same ordering logic as built-in modules, and custom modules
+are appended after built-ins before ordering:
+
+```python
+from mvt.common.module import MVTModule
+from mvt.ios.modules.backup.manifest import Manifest
+
+
+class DependentCustomModule(MVTModule):
+    supported_commands = (("ios", "check-backup"),)
+    dependencies = (Manifest,)
+
+    def run(self):
+        manifest_results = self.get_dependency_results(Manifest)
+        self.results = [{"manifest_entries": len(manifest_results)}]
+```
+
 ## 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

docs/iocs.md

@@ -34,6 +34,25 @@ It is also possible to load STIX2 files automatically from the environment varia
 export MVT_STIX2="/home/user/IOC1.stix2:/home/user/IOC2.stix2"
 ```
 
+## Network Access
+
+When checking URL indicators, MVT follows recognized shortened URLs with an
+HTTP `HEAD` request. The following environment variables control these
+requests:
+
+- `MVT_NETWORK_ACCESS_ALLOWED` enables or disables network requests. It defaults
+  to `true`. Set it to `false` to prevent MVT from attempting to resolve
+  shortened URLs.
+- `MVT_NETWORK_TIMEOUT` sets the request timeout in seconds. It defaults to
+  `15`.
+
+For example, to run IOC checks without resolving shortened URLs:
+
+```bash
+MVT_NETWORK_ACCESS_ALLOWED=false mvt-ios check-iocs \
+    --iocs ~/iocs/malware.stix2 /path/to/iphone/output/
+```
+
 ## STIX2 Support
 
 So far MVT implements only a subset of [STIX2 specifications](https://docs.oasis-open.org/cti/stix/v2.1/csprd01/stix-v2.1-csprd01.html):
@@ -55,4 +74,3 @@ You can automaticallly download the latest public indicator files with the comma
 Please [open an issue](https://github.com/mvt-project/mvt/issues/) to suggest new sources of STIX-formatted IOCs.
 
 
-

pyproject.toml

@@ -17,22 +17,22 @@ classifiers = [
     "Programming Language :: Python",
 ]
 dependencies = [
-    "click==8.3.2",
+    "click==8.3.3",
     "rich==14.3.3",
-    "tld==0.13.1",
+    "tld==0.13.2",
     "requests==2.33.1",
     "simplejson==3.20.2",
-    "packaging==26.0",
+    "packaging==26.2",
     "appdirs==1.4.4",
     "iOSbackup==0.9.925",
     "adb-shell[usb]==0.4.4",
     "libusb1==3.3.1",
     "cryptography==47.0.0",
     "PyYAML>=6.0.2",
-    "pyahocorasick==2.2.0",
+    "pyahocorasick==2.3.1",
     "betterproto2==0.9.1",
     "pydantic==2.13.3",
-    "pydantic-settings==2.13.1",
+    "pydantic-settings==2.14.0",
     "NSKeyedUnArchiver==1.5.2",
     "python-dateutil==2.9.0.post0",
     "tzdata==2026.2",

src/mvt/android/artifacts/dumpsys_adb.py

@@ -131,10 +131,17 @@ class DumpsysADBArtifact(AndroidArtifact):
             )
             return
 
-        # TODO: Parse AdbDebuggingManager line in output.
-        start_of_json = content.find(b"\n{") + 2
-        end_of_json = content.rfind(b"}\n") - 2
-        json_content = content[start_of_json:end_of_json].rstrip()
+        start_of_json = content.find(b"\n{")
+        if start_of_json == -1:
+            self.log.error("Unable to find ADB manager state in dumpsys output")
+            return
+
+        end_of_json = content.rfind(b"}\n")
+        if end_of_json == -1 or end_of_json <= start_of_json:
+            self.log.error("Unable to find complete ADB manager state in dumpsys output")
+            return
+
+        json_content = content[start_of_json + 2 : end_of_json - 2].rstrip()
 
         parsed = self.indented_dump_parser(json_content)
         if parsed.get("debugging_manager") is None:

src/mvt/android/artifacts/dumpsys_battery_daily.py

@@ -50,13 +50,11 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact):
     def parse(self, output: str) -> None:
         daily = None
         daily_updates: list[dict[str, Any]] = []
-        package_versions: dict[
-            str, str
-        ] = {}  # Track package versions to detect downgrades
+        records: list[dict[str, Any]] = []
         for line in output.splitlines():
             if line.startswith("  Daily from "):
                 if len(daily_updates) > 0:
-                    self.results.extend(daily_updates)
+                    records.extend(daily_updates)
                     daily_updates = []
 
                 timeframe = line[13:].strip()
@@ -89,35 +87,53 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact):
                     "vers": vers_nr,
                 }
 
-                # Check for uninstall (version 0)
-                if vers_nr == "0":
-                    self.alertstore.medium(
-                        f"Detected uninstall of package {package_name} (vers 0)",
-                        daily["from"],
-                        update_record,
-                    )
-                # Check for downgrade
-                elif package_name in package_versions:
-                    try:
-                        current_vers = int(vers_nr)
-                        previous_vers = int(package_versions[package_name])
-                        if current_vers < previous_vers:
-                            update_record["action"] = "downgrade"
-                            update_record["previous_vers"] = str(previous_vers)
-                            self.alertstore.medium(
-                                f"Detected downgrade of package {package_name} "
-                                f"from vers {previous_vers} to vers {current_vers}",
-                                daily["from"],
-                                update_record,
-                            )
-                    except ValueError:
-                        # If version numbers aren't integers, skip comparison
-                        pass
-
-                # Update tracking dictionary
-                package_versions[package_name] = vers_nr
-
                 daily_updates.append(update_record)
 
         if len(daily_updates) > 0:
-            self.results.extend(daily_updates)
+            records.extend(daily_updates)
+
+        self._detect_uninstalls_and_downgrades(records)
+        self.results.extend(records)
+
+    def _detect_uninstalls_and_downgrades(
+        self, records: list[dict[str, Any]]
+    ) -> None:
+        package_versions: dict[str, int] = {}
+
+        for record in sorted(
+            records,
+            key=lambda record: (
+                record["from"],
+                record["to"],
+                record["package_name"],
+            ),
+        ):
+            package_name = record["package_name"]
+            vers_nr = record["vers"]
+
+            if vers_nr == "0":
+                self.alertstore.medium(
+                    f"Detected uninstall of package {package_name} (vers 0)",
+                    record["from"],
+                    record,
+                )
+                package_versions.pop(package_name, None)
+                continue
+
+            try:
+                current_vers = int(vers_nr)
+            except ValueError:
+                continue
+
+            previous_vers = package_versions.get(package_name)
+            if previous_vers is not None and current_vers < previous_vers:
+                record["action"] = "downgrade"
+                record["previous_vers"] = str(previous_vers)
+                self.alertstore.medium(
+                    f"Detected downgrade of package {package_name} "
+                    f"from vers {previous_vers} to vers {current_vers}",
+                    record["from"],
+                    record,
+                )
+
+            package_versions[package_name] = current_vers

src/mvt/android/artifacts/dumpsys_packages.py

@@ -14,9 +14,12 @@ from .artifact import AndroidArtifact
 
 class DumpsysPackagesArtifact(AndroidArtifact):
     def check_indicators(self) -> None:
+        alerted_root_packages = set()
         for result in self.results:
-            # XXX: De-duplication Package detections
             if result["package_name"] in ROOT_PACKAGES:
+                if result["package_name"] in alerted_root_packages:
+                    continue
+                alerted_root_packages.add(result["package_name"])
                 self.alertstore.medium(
                     f'Found an installed package related to rooting/jailbreaking: "{result["package_name"]}"',
                     "",
@@ -188,7 +191,7 @@ class DumpsysPackagesArtifact(AndroidArtifact):
         package = []
 
         in_package_list = False
-        for line in content.split("\n"):
+        for line in content.splitlines():
             if line.startswith("Packages:"):
                 in_package_list = True
                 continue

src/mvt/android/artifacts/processes.py

@@ -8,7 +8,7 @@ from .artifact import AndroidArtifact
 
 class Processes(AndroidArtifact):
     def parse(self, entry: str) -> None:
-        for line in entry.split("\n")[1:]:
+        for line in entry.splitlines()[1:]:
             proc = line.split()
 
             # Skip empty lines

src/mvt/android/artifacts/settings.py

@@ -67,11 +67,14 @@ class Settings(AndroidArtifact):
                     # Check if one of the dangerous settings is using an unsafe
                     # value (different than the one specified).
                     if danger["key"] == key and danger["safe_value"] != value:
-                        self.log.warning(
-                            'Found suspicious "%s" setting "%s = %s" (%s)',
-                            namespace,
-                            key,
-                            value,
-                            danger["description"],
+                        self.alertstore.medium(
+                            f'Found suspicious "{namespace}" setting "{key} = {value}" ({danger["description"]})',
+                            "",
+                            {
+                                "namespace": namespace,
+                                "key": key,
+                                "value": value,
+                                "description": danger["description"],
+                            },
                         )
                         break

src/mvt/android/artifacts/tombstone_crashes.py

@@ -101,8 +101,7 @@ class TombstoneCrashArtifact(AndroidArtifact):
                 continue
 
             if result.get("command_line", []):
-                command_name = result.get("command_line")[0].split("/")[-1]
-                command_name = result["command_line"][0]
+                command_name = result["command_line"][0].split("/")[-1]
                 ioc_match = self.indicators.check_process(command_name)
                 if ioc_match:
                     self.alertstore.critical(
@@ -200,7 +199,7 @@ class TombstoneCrashArtifact(AndroidArtifact):
             # eg. "Process uptime: 40s"
             tombstone[destination_key] = int(value_clean.rstrip("s"))
         elif destination_key == "command_line":
-            # XXX: Check if command line should be a single string in a list, or a list of strings.
+            # Wrap in list for consistency with protobuf format (repeated string).
             tombstone[destination_key] = [value_clean]
         else:
             tombstone[destination_key] = value_clean
@@ -262,7 +261,7 @@ class TombstoneCrashArtifact(AndroidArtifact):
     @staticmethod
     def _parse_timestamp_string(timestamp: str) -> str:
         timestamp_parsed = parser.parse(timestamp)
-        # HACK: Swap the local timestamp to UTC, so keep the original time and avoid timezone conversion.
+        # Preserve the source wall-clock time while returning the project-wide ISO format.
         local_timestamp = timestamp_parsed.replace(tzinfo=datetime.timezone.utc)
         return convert_datetime_to_iso(local_timestamp)
 

src/mvt/android/cli.py

@@ -8,6 +8,12 @@ import logging
 import click
 
 from mvt.common.cmd_check_iocs import CmdCheckIOCS
+from mvt.common.completion import (
+    SUPPORTED_SHELLS,
+    completion_instructions,
+    generate_completion_script,
+    install_completion_script,
+)
 from mvt.common.help import (
     HELP_MSG_ANDROID_BACKUP_PASSWORD,
     HELP_MSG_CHECK_ADB_REMOVED,
@@ -17,11 +23,13 @@ from mvt.common.help import (
     HELP_MSG_CHECK_BUGREPORT,
     HELP_MSG_CHECK_IOCS,
     HELP_MSG_CHECK_INTRUSION_LOGS,
+    HELP_MSG_COMPLETION,
     HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK,
     HELP_MSG_DISABLE_UPDATE_CHECK,
     HELP_MSG_HASHES,
     HELP_MSG_IOC,
     HELP_MSG_LIST_MODULES,
+    HELP_MSG_LOAD_MODULE,
     HELP_MSG_MODULE,
     HELP_MSG_NONINTERACTIVE,
     HELP_MSG_OUTPUT,
@@ -30,6 +38,7 @@ from mvt.common.help import (
     HELP_MSG_VERSION,
 )
 from mvt.common.logo import logo
+from mvt.common.module_loader import CustomModuleLoadError, load_custom_modules
 from mvt.common.updates import IndicatorsUpdates
 from mvt.common.utils import init_logging, set_verbose_logging
 
@@ -59,6 +68,13 @@ def _get_disable_flags(ctx):
     )
 
 
+def _load_custom_modules(load_module):
+    try:
+        return load_custom_modules(load_module)
+    except CustomModuleLoadError as exc:
+        raise click.ClickException(str(exc)) from exc
+
+
 # ==============================================================================
 # Main
 # ==============================================================================
@@ -76,10 +92,11 @@ def cli(ctx, disable_update_check, disable_indicator_update_check):
     ctx.ensure_object(dict)
     ctx.obj["disable_version_check"] = disable_update_check
     ctx.obj["disable_indicator_check"] = disable_indicator_update_check
-    logo(
-        disable_version_check=disable_update_check,
-        disable_indicator_check=disable_indicator_update_check,
-    )
+    if ctx.invoked_subcommand != "completion":
+        logo(
+            disable_version_check=disable_update_check,
+            disable_indicator_check=disable_indicator_update_check,
+        )
 
 
 # ==============================================================================
@@ -90,6 +107,40 @@ def version():
     return
 
 
+# ==============================================================================
+# Command: completion
+# ==============================================================================
+@cli.command("completion", context_settings=CONTEXT_SETTINGS, help=HELP_MSG_COMPLETION)
+@click.argument("shell", required=False, type=click.Choice(SUPPORTED_SHELLS))
+@click.option(
+    "--install",
+    is_flag=True,
+    help="Write completion files and update shell configuration.",
+)
+@click.pass_context
+def completion(ctx, shell, install):
+    program_name = "mvt-android"
+
+    if shell is None:
+        if install:
+            raise click.UsageError("A shell is required when using --install.")
+        click.echo(completion_instructions(program_name))
+        return
+
+    root_cli = ctx.find_root().command
+
+    if install:
+        script_path = install_completion_script(root_cli, program_name, shell)
+        click.echo(f"Installed {shell} completion to {script_path}")
+        if shell in ("bash", "zsh"):
+            click.echo(f"Updated ~/.{shell}rc")
+        else:
+            click.echo("Fish loads completion files automatically.")
+        return
+
+    click.echo(generate_completion_script(root_cli, program_name, shell))
+
+
 # ==============================================================================
 # Command: check-adb (removed)
 # ==============================================================================
@@ -119,11 +170,28 @@ def check_adb(ctx):
 @click.option("--output", "-o", type=click.Path(exists=False), help=HELP_MSG_OUTPUT)
 @click.option("--list-modules", "-l", is_flag=True, help=HELP_MSG_LIST_MODULES)
 @click.option("--module", "-m", help=HELP_MSG_MODULE)
+@click.option(
+    "--load-module",
+    type=click.Path(exists=True),
+    multiple=True,
+    default=[],
+    help=HELP_MSG_LOAD_MODULE,
+)
 @click.option("--verbose", "-v", is_flag=True, help=HELP_MSG_VERBOSE)
 @click.argument("BUGREPORT_PATH", type=click.Path(exists=True))
 @click.pass_context
-def check_bugreport(ctx, iocs, output, list_modules, module, verbose, bugreport_path):
+def check_bugreport(
+    ctx,
+    iocs,
+    output,
+    list_modules,
+    module,
+    load_module,
+    verbose,
+    bugreport_path,
+):
     set_verbose_logging(verbose)
+    custom_modules = _load_custom_modules(load_module)
     # Always generate hashes as bug reports are small.
     cmd = CmdAndroidCheckBugreport(
         target_path=bugreport_path,
@@ -133,6 +201,7 @@ def check_bugreport(ctx, iocs, output, list_modules, module, verbose, bugreport_
         hashes=True,
         disable_version_check=_get_disable_flags(ctx)[0],
         disable_indicator_check=_get_disable_flags(ctx)[1],
+        custom_modules=custom_modules,
     )
 
     if list_modules:
@@ -164,6 +233,13 @@ def check_bugreport(ctx, iocs, output, list_modules, module, verbose, bugreport_
 )
 @click.option("--output", "-o", type=click.Path(exists=False), help=HELP_MSG_OUTPUT)
 @click.option("--list-modules", "-l", is_flag=True, help=HELP_MSG_LIST_MODULES)
+@click.option(
+    "--load-module",
+    type=click.Path(exists=True),
+    multiple=True,
+    default=[],
+    help=HELP_MSG_LOAD_MODULE,
+)
 @click.option("--non-interactive", "-n", is_flag=True, help=HELP_MSG_NONINTERACTIVE)
 @click.option("--backup-password", "-p", help=HELP_MSG_ANDROID_BACKUP_PASSWORD)
 @click.option("--verbose", "-v", is_flag=True, help=HELP_MSG_VERBOSE)
@@ -174,12 +250,14 @@ def check_backup(
     iocs,
     output,
     list_modules,
+    load_module,
     non_interactive,
     backup_password,
     verbose,
     backup_path,
 ):
     set_verbose_logging(verbose)
+    custom_modules = _load_custom_modules(load_module)
 
     # Always generate hashes as backups are generally small.
     cmd = CmdAndroidCheckBackup(
@@ -193,6 +271,7 @@ def check_backup(
         },
         disable_version_check=_get_disable_flags(ctx)[0],
         disable_indicator_check=_get_disable_flags(ctx)[1],
+        custom_modules=custom_modules,
     )
 
     if list_modules:
@@ -223,6 +302,13 @@ def check_backup(
 @click.option("--output", "-o", type=click.Path(exists=False), help=HELP_MSG_OUTPUT)
 @click.option("--list-modules", "-l", is_flag=True, help=HELP_MSG_LIST_MODULES)
 @click.option("--module", "-m", help=HELP_MSG_MODULE)
+@click.option(
+    "--load-module",
+    type=click.Path(exists=True),
+    multiple=True,
+    default=[],
+    help=HELP_MSG_LOAD_MODULE,
+)
 @click.option("--hashes", "-H", is_flag=True, help=HELP_MSG_HASHES)
 @click.option("--non-interactive", "-n", is_flag=True, help=HELP_MSG_NONINTERACTIVE)
 @click.option("--backup-password", "-p", help=HELP_MSG_ANDROID_BACKUP_PASSWORD)
@@ -235,6 +321,7 @@ def check_androidqf(
     output,
     list_modules,
     module,
+    load_module,
     hashes,
     non_interactive,
     backup_password,
@@ -242,6 +329,7 @@ def check_androidqf(
     androidqf_path,
 ):
     set_verbose_logging(verbose)
+    custom_modules = _load_custom_modules(load_module)
 
     cmd = CmdAndroidCheckAndroidQF(
         target_path=androidqf_path,
@@ -255,6 +343,7 @@ def check_androidqf(
         },
         disable_version_check=_get_disable_flags(ctx)[0],
         disable_indicator_check=_get_disable_flags(ctx)[1],
+        custom_modules=custom_modules,
     )
 
     if list_modules:
@@ -288,6 +377,13 @@ def check_androidqf(
 @click.option("--output", "-o", type=click.Path(exists=False), help=HELP_MSG_OUTPUT)
 @click.option("--list-modules", "-l", is_flag=True, help=HELP_MSG_LIST_MODULES)
 @click.option("--module", "-m", help=HELP_MSG_MODULE)
+@click.option(
+    "--load-module",
+    type=click.Path(exists=True),
+    multiple=True,
+    default=[],
+    help=HELP_MSG_LOAD_MODULE,
+)
 @click.option(
     "--timezone",
     "-t",
@@ -307,11 +403,13 @@ def check_intrusion_logs(
     output,
     list_modules,
     module,
+    load_module,
     timezone,
     verbose,
     logs_path,
 ):
     set_verbose_logging(verbose)
+    custom_modules = _load_custom_modules(load_module)
 
     module_options = {}
     if timezone:
@@ -325,6 +423,7 @@ def check_intrusion_logs(
         module_options=module_options if module_options else None,
         disable_version_check=_get_disable_flags(ctx)[0],
         disable_indicator_check=_get_disable_flags(ctx)[1],
+        custom_modules=custom_modules,
     )
 
     if list_modules:
@@ -352,10 +451,26 @@ def check_intrusion_logs(
 )
 @click.option("--list-modules", "-l", is_flag=True, help=HELP_MSG_LIST_MODULES)
 @click.option("--module", "-m", help=HELP_MSG_MODULE)
+@click.option(
+    "--load-module",
+    type=click.Path(exists=True),
+    multiple=True,
+    default=[],
+    help=HELP_MSG_LOAD_MODULE,
+)
 @click.argument("FOLDER", type=click.Path(exists=True))
 @click.pass_context
-def check_iocs(ctx, iocs, list_modules, module, folder):
-    cmd = CmdCheckIOCS(target_path=folder, ioc_files=iocs, module_name=module)
+def check_iocs(ctx, iocs, list_modules, module, load_module, folder):
+    custom_modules = _load_custom_modules(load_module)
+    cmd = CmdCheckIOCS(
+        target_path=folder,
+        ioc_files=iocs,
+        module_name=module,
+        disable_version_check=_get_disable_flags(ctx)[0],
+        disable_indicator_check=_get_disable_flags(ctx)[1],
+        custom_modules=custom_modules,
+        platform="android",
+    )
     cmd.modules = (
         BACKUP_MODULES + BUGREPORT_MODULES + ANDROIDQF_MODULES + INTRUSION_LOGS_MODULES
     )

src/mvt/android/cmd_check_androidqf.py

@@ -13,10 +13,11 @@ from typing import List, Optional
 
 from mvt.android.artifacts.getprop import GetProp
 from mvt.android.cmd_check_intrusion_logs import CmdAndroidCheckIntrusionLogs
-from mvt.android.cmd_check_backup import CmdAndroidCheckBackup
+from mvt.android.cmd_check_backup import CmdAndroidCheckBackup, InvalidAndroidBackup
 from mvt.android.cmd_check_bugreport import CmdAndroidCheckBugreport
 from mvt.common.command import Command
 from mvt.common.indicators import Indicators
+from mvt.common.module import MVTModule
 
 from .modules.androidqf import ANDROIDQF_MODULES
 from .modules.androidqf.base import AndroidQFModule
@@ -50,6 +51,7 @@ class CmdAndroidCheckAndroidQF(Command):
         sub_command: Optional[bool] = False,
         disable_version_check: bool = False,
         disable_indicator_check: bool = False,
+        custom_modules: Optional[list[type[MVTModule]]] = None,
     ) -> None:
         super().__init__(
             target_path=target_path,
@@ -64,8 +66,10 @@ class CmdAndroidCheckAndroidQF(Command):
             log=log,
             disable_version_check=disable_version_check,
             disable_indicator_check=disable_indicator_check,
+            custom_modules=custom_modules,
         )
 
+        self.platform = "android"
         self.name = "check-androidqf"
         self.modules = ANDROIDQF_MODULES
 
@@ -210,6 +214,7 @@ class CmdAndroidCheckAndroidQF(Command):
                 module_options=self.module_options,
                 hashes=self.hashes,
                 sub_command=True,
+                custom_modules=self.custom_modules,
             )
             cmd.from_zip(bugreport)
             cmd.run()
@@ -239,8 +244,16 @@ class CmdAndroidCheckAndroidQF(Command):
             module_options=self.module_options,
             hashes=self.hashes,
             sub_command=True,
+            custom_modules=self.custom_modules,
         )
-        cmd.from_ab(backup)
+        try:
+            cmd.from_ab(backup)
+        except InvalidAndroidBackup as exc:
+            self.log.warning(
+                "Skipping backup modules as backup.ab is malformed: %s", exc
+            )
+            return False
+
         cmd.run()
 
         self.timeline.extend(cmd.timeline)
@@ -311,6 +324,7 @@ class CmdAndroidCheckAndroidQF(Command):
                 module_options=adv_module_options,
                 hashes=self.hashes,
                 sub_command=True,
+                custom_modules=self.custom_modules,
             )
             cmd.run()
 

src/mvt/android/cmd_check_backup.py

@@ -21,12 +21,17 @@ from mvt.android.parsers.backup import (
 )
 from mvt.common.command import Command
 from mvt.common.indicators import Indicators
+from mvt.common.module import MVTModule
 
 from .modules.backup import BACKUP_MODULES
 
 log = logging.getLogger(__name__)
 
 
+class InvalidAndroidBackup(Exception):
+    pass
+
+
 class CmdAndroidCheckBackup(Command):
     def __init__(
         self,
@@ -41,6 +46,7 @@ class CmdAndroidCheckBackup(Command):
         sub_command: Optional[bool] = False,
         disable_version_check: bool = False,
         disable_indicator_check: bool = False,
+        custom_modules: Optional[list[type[MVTModule]]] = None,
     ) -> None:
         super().__init__(
             target_path=target_path,
@@ -55,8 +61,10 @@ class CmdAndroidCheckBackup(Command):
             log=log,
             disable_version_check=disable_version_check,
             disable_indicator_check=disable_indicator_check,
+            custom_modules=custom_modules,
         )
 
+        self.platform = "android"
         self.name = "check-backup"
         self.modules = BACKUP_MODULES
 
@@ -68,6 +76,10 @@ class CmdAndroidCheckBackup(Command):
         self.__type = "ab"
         header = parse_ab_header(ab_file_bytes)
         if not header["backup"]:
+            if self.sub_command:
+                raise InvalidAndroidBackup(
+                    "Invalid backup format, file should be in .ab format"
+                )
             log.critical("Invalid backup format, file should be in .ab format")
             sys.exit(1)
 
@@ -83,12 +95,25 @@ class CmdAndroidCheckBackup(Command):
             log.critical("Invalid backup password")
             sys.exit(1)
         except AndroidBackupParsingError as exc:
+            if self.sub_command:
+                raise InvalidAndroidBackup(
+                    f"Impossible to parse this backup file: {exc}"
+                ) from exc
             log.critical("Impossible to parse this backup file: %s", exc)
             log.critical("Please use Android Backup Extractor (ABE) instead")
             sys.exit(1)
 
         dbytes = io.BytesIO(tardata)
-        self.__tar = tarfile.open(fileobj=dbytes)
+        try:
+            self.__tar = tarfile.open(fileobj=dbytes)
+        except tarfile.TarError as exc:
+            if self.sub_command:
+                raise InvalidAndroidBackup(
+                    f"Impossible to parse this backup file: {exc}"
+                ) from exc
+            log.critical("Impossible to parse this backup file: %s", exc)
+            log.critical("Please use Android Backup Extractor (ABE) instead")
+            sys.exit(1)
         for member in self.__tar:
             self.__files.append(member.name)
 

src/mvt/android/cmd_check_bugreport.py

@@ -12,6 +12,7 @@ from zipfile import ZipFile
 from mvt.android.modules.bugreport.base import BugReportModule
 from mvt.common.command import Command
 from mvt.common.indicators import Indicators
+from mvt.common.module import MVTModule
 
 from .modules.bugreport import BUGREPORT_MODULES
 
@@ -32,6 +33,7 @@ class CmdAndroidCheckBugreport(Command):
         sub_command: Optional[bool] = False,
         disable_version_check: bool = False,
         disable_indicator_check: bool = False,
+        custom_modules: Optional[list[type[MVTModule]]] = None,
     ) -> None:
         super().__init__(
             target_path=target_path,
@@ -46,8 +48,10 @@ class CmdAndroidCheckBugreport(Command):
             log=log,
             disable_version_check=disable_version_check,
             disable_indicator_check=disable_indicator_check,
+            custom_modules=custom_modules,
         )
 
+        self.platform = "android"
         self.name = "check-bugreport"
         self.modules = BUGREPORT_MODULES
 

src/mvt/android/cmd_check_intrusion_logs.py

@@ -9,6 +9,7 @@ from typing import Optional
 
 from mvt.common.command import Command
 from mvt.common.indicators import Indicators
+from mvt.common.module import MVTModule
 
 from .modules.intrusion_logs import (
     INTRUSION_LOGS_MODULES,
@@ -35,6 +36,7 @@ class CmdAndroidCheckIntrusionLogs(Command):
         sub_command: Optional[bool] = False,
         disable_version_check: bool = False,
         disable_indicator_check: bool = False,
+        custom_modules: Optional[list[type[MVTModule]]] = None,
     ) -> None:
         super().__init__(
             target_path=target_path,
@@ -49,8 +51,10 @@ class CmdAndroidCheckIntrusionLogs(Command):
             log=log,
             disable_version_check=disable_version_check,
             disable_indicator_check=disable_indicator_check,
+            custom_modules=custom_modules,
         )
 
+        self.platform = "android"
         self.name = "check-intrusion-logs"
         self.modules = INTRUSION_LOGS_MODULES
         self._all_events: dict[str, list[dict]] = {}

src/mvt/android/modules/androidqf/aqf_files.py

@@ -41,7 +41,7 @@ class AQFFiles(AndroidQFModule):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,
@@ -107,16 +107,16 @@ class AQFFiles(AndroidQFModule):
                     msg = f'Found {file_type}file at suspicious path "{result["path"]}"'
                     self.alertstore.high(msg, "", result)
 
-            if result.get("sha256", "") == "":
-                continue
-
-            ioc_match = self.indicators.check_file_hash(result.get("sha256") or "")
-            if ioc_match:
-                self.alertstore.critical(
-                    ioc_match.message, "", result, matched_indicator=ioc_match.ioc
-                )
-
-            # TODO: adds SHA1 and MD5 when available in MVT
+            for hash_key in ("sha256", "sha1", "md5"):
+                file_hash = result.get(hash_key, "")
+                if not file_hash:
+                    continue
+                ioc_match = self.indicators.check_file_hash(file_hash)
+                if ioc_match:
+                    self.alertstore.critical(
+                        ioc_match.message, "", result, matched_indicator=ioc_match.ioc
+                    )
+                    break
 
     def run(self) -> None:
         if timezone := self._get_device_timezone():
@@ -131,7 +131,7 @@ class AQFFiles(AndroidQFModule):
                 data = json.loads(rawdata)
             except json.decoder.JSONDecodeError:
                 data = []
-                for line in rawdata.split("\n"):
+                for line in rawdata.splitlines():
                     if line.strip() == "":
                         continue
                     data.append(json.loads(line))
@@ -142,11 +142,11 @@ class AQFFiles(AndroidQFModule):
                         utc_timestamp = datetime.datetime.fromtimestamp(
                             file_data[ts], tz=datetime.timezone.utc
                         )
-                        # Convert the UTC timestamp to local tiem on Android device's local timezone
+                        # Convert the UTC timestamp to local time on Android device's local timezone
                         local_timestamp = utc_timestamp.astimezone(device_timezone)
 
-                        # HACK: We only output the UTC timestamp in convert_datetime_to_iso, we
-                        # set the timestamp timezone to UTC, to avoid the timezone conversion again.
+                        # Preserve the device-local wall-clock time while using
+                        # the project-wide ISO conversion helper.
                         local_timestamp = local_timestamp.replace(
                             tzinfo=datetime.timezone.utc
                         )

src/mvt/android/modules/androidqf/aqf_getprop.py

@@ -22,7 +22,7 @@ class AQFGetProp(GetPropArtifact, AndroidQFModule):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,
@@ -32,7 +32,7 @@ class AQFGetProp(GetPropArtifact, AndroidQFModule):
             log=log,
             results=results,
         )
-        self.results: list = []
+        self.results: list = [] if results is None else results
 
     def run(self) -> None:
         getprop_files = self._get_files_by_pattern("*/getprop.txt")

src/mvt/android/modules/androidqf/aqf_log_timestamps.py

@@ -27,7 +27,7 @@ class AQFLogTimestamps(FileTimestampsArtifact, AndroidQFModule):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/androidqf/aqf_packages.py

@@ -30,7 +30,7 @@ class AQFPackages(AndroidQFModule):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/androidqf/aqf_processes.py

@@ -22,7 +22,7 @@ class AQFProcesses(ProcessesArtifact, AndroidQFModule):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/androidqf/aqf_settings.py

@@ -22,7 +22,7 @@ class AQFSettings(SettingsArtifact, AndroidQFModule):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,
@@ -32,7 +32,7 @@ class AQFSettings(SettingsArtifact, AndroidQFModule):
             log=log,
             results=results,
         )
-        self.results: dict = {}
+        self.results: dict = results if results is not None else {}
 
     def run(self) -> None:
         for setting_file in self._get_files_by_pattern("*/settings_*.txt"):
@@ -40,7 +40,7 @@ class AQFSettings(SettingsArtifact, AndroidQFModule):
 
             self.results[namespace] = {}
             data = self._get_file_content(setting_file)
-            for line in data.decode("utf-8").split("\n"):
+            for line in data.decode("utf-8").splitlines():
                 line = line.strip()
                 try:
                     key, value = line.split("=", 1)

src/mvt/android/modules/androidqf/base.py

@@ -23,7 +23,7 @@ class AndroidQFModule(MVTModule):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/androidqf/mounts.py

@@ -32,7 +32,7 @@ class Mounts(MountsArtifact, AndroidQFModule):
             log=log,
             results=results,
         )
-        self.results: list = []
+        self.results: list = [] if results is None else results
 
     def run(self) -> None:
         """
@@ -66,6 +66,9 @@ class Mounts(MountsArtifact, AndroidQFModule):
                 # AndroidQF format: array of strings like
                 # "/dev/block/dm-12 on / type ext4 (ro,seclabel,noatime)"
                 mount_content = "\n".join(json_data)
+            else:
+                self.log.error("Expected mounts.json to contain a list of mount lines")
+                return
             self.parse(mount_content)
         except Exception as exc:
             self.log.error("Failed to parse mount information: %s", exc)

src/mvt/android/modules/androidqf/sms.py

@@ -21,10 +21,6 @@ from .base import AndroidQFModule
 class SMS(AndroidQFModule):
     """
     This module analyse SMS file in backup
-
-    XXX: We should also de-duplicate this AQF module, but first we
-    need to add tests for loading encrypted SMS backups through the backup
-    sub-module.
     """
 
     def __init__(
@@ -62,7 +58,7 @@ class SMS(AndroidQFModule):
     def parse_backup(self, data):
         header = parse_ab_header(data)
         if not header["backup"]:
-            self.log.critical("Invalid backup format, backup.ab was not analysed")
+            self.log.warning("Invalid backup format, backup.ab was not analysed")
             return
 
         password = None
@@ -80,7 +76,7 @@ class SMS(AndroidQFModule):
             self.log.critical("Invalid backup password")
             return
         except AndroidBackupParsingError:
-            self.log.critical(
+            self.log.warning(
                 "Impossible to parse this backup file, please use"
                 " Android Backup Extractor instead"
             )

src/mvt/android/modules/backup/base.py

@@ -22,7 +22,7 @@ class BackupModule(MVTModule):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/backup/sms.py

@@ -20,7 +20,7 @@ class SMS(BackupModule):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/bugreport/base.py

@@ -23,7 +23,7 @@ class BugReportModule(MVTModule):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/bugreport/dumpsys_accessibility.py

@@ -22,7 +22,7 @@ class DumpsysAccessibility(DumpsysAccessibilityArtifact, BugReportModule):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/bugreport/dumpsys_activities.py

@@ -24,7 +24,7 @@ class DumpsysActivities(DumpsysPackageActivitiesArtifact, BugReportModule):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/bugreport/dumpsys_adb_state.py

@@ -22,7 +22,7 @@ class DumpsysADBState(DumpsysADBArtifact, BugReportModule):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/bugreport/dumpsys_appops.py

@@ -22,7 +22,7 @@ class DumpsysAppops(DumpsysAppopsArtifact, BugReportModule):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/bugreport/dumpsys_battery_daily.py

@@ -22,7 +22,7 @@ class DumpsysBatteryDaily(DumpsysBatteryDailyArtifact, BugReportModule):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/bugreport/dumpsys_battery_history.py

@@ -22,7 +22,7 @@ class DumpsysBatteryHistory(DumpsysBatteryHistoryArtifact, BugReportModule):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/bugreport/dumpsys_dbinfo.py

@@ -24,7 +24,7 @@ class DumpsysDBInfo(DumpsysDBInfoArtifact, BugReportModule):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/bugreport/dumpsys_getprop.py

@@ -22,7 +22,7 @@ class DumpsysGetProp(GetPropArtifact, BugReportModule):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/bugreport/dumpsys_packages.py

@@ -23,7 +23,7 @@ class DumpsysPackages(DumpsysPackagesArtifact, BugReportModule):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/bugreport/dumpsys_platform_compat.py

@@ -22,7 +22,7 @@ class DumpsysPlatformCompat(DumpsysPlatformCompatArtifact, BugReportModule):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,
@@ -42,8 +42,10 @@ class DumpsysPlatformCompat(DumpsysPlatformCompatArtifact, BugReportModule):
             )
             return
 
-        data = data.decode("utf-8", errors="replace")
-        content = self.extract_dumpsys_section(data, "DUMP OF SERVICE platform_compat:")
+        decoded_data = data.decode("utf-8", errors="replace")
+        content = self.extract_dumpsys_section(
+            decoded_data, "DUMP OF SERVICE platform_compat:"
+        )
         self.parse(content)
 
         self.log.info("Found %d uninstalled apps", len(self.results))

src/mvt/android/modules/bugreport/dumpsys_receivers.py

@@ -22,7 +22,7 @@ class DumpsysReceivers(DumpsysReceiversArtifact, BugReportModule):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/bugreport/fs_timestamps.py

@@ -24,7 +24,7 @@ class BugReportTimestamps(FileTimestampsArtifact, BugReportModule):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/bugreport/tombstones.py

@@ -23,7 +23,7 @@ class Tombstones(TombstoneCrashArtifact, BugReportModule):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/android/modules/intrusion_logs/security_event.py

@@ -264,6 +264,7 @@ SECURITY_EVENT_TAGS = {
 }
 
 SECURITY_EVENT_METADATA_KEYS = {
+    "event_id",
     "event_time",
     "event_type",
     "timestamp",
@@ -302,10 +303,15 @@ class SecurityEvent(IntrusionLogsModule):
 
     def check_indicators(self) -> None:
         """Check security events against indicators of compromise."""
-        if not self.indicators:
-            return
-
         for result in self.results:
+            # Heuristic alerts are intrinsic to the event, so they run even
+            # when no indicator set is loaded.
+            self._check_security_heuristics(result)
+
+            # The remaining checks match events against loaded indicators.
+            if not self.indicators:
+                continue
+
             # Check app process start events for suspicious package names
             if "app_process_start" in result:
                 process_info = result["app_process_start"]
@@ -389,60 +395,73 @@ class SecurityEvent(IntrusionLogsModule):
                                 matched_indicator=ioc.ioc,
                             )
 
-            # Flag failed cryptographic operations as potentially suspicious
-            if "key_generated" in result:
-                if not result["key_generated"].get("success", True):
-                    self.log.warning(
-                        "Failed key generation detected for key_id: %s",
-                        result["key_generated"].get("key_id", "unknown"),
-                    )
-
-            # Flag certificate validation failures
-            if "cert_validation_failure" in result:
+    def _check_security_heuristics(self, result: dict) -> None:
+        """Raise alerts for events that are intrinsically suspicious,
+        independent of any loaded indicators."""
+        # Flag failed cryptographic operations as potentially suspicious
+        if "key_generated" in result:
+            if not result["key_generated"].get("success", True):
                 self.log.warning(
-                    "Certificate validation failure detected: %s",
-                    result.get("cert_validation_failure"),
+                    "Failed key generation detected for key_id: %s",
+                    result["key_generated"].get("key_id", "unknown"),
                 )
 
-            # Flag key integrity violations
-            if "key_integrity_violation" in result:
+        # Flag certificate validation failures
+        if "cert_validation_failure" in result:
+            self.alertstore.medium(
+                "Certificate validation failure detected: "
+                f"{result.get('cert_validation_failure')}",
+                result.get("timestamp") or "",
+                result,
+            )
+
+        # Flag key integrity violations
+        if "key_integrity_violation" in result:
+            self.alertstore.medium(
+                f"Key integrity violation detected: {result.get('key_integrity_violation')}",
+                result.get("timestamp") or "",
+                result,
+            )
+
+        # Flag successful certificate authority installations (potential
+        # MITM); a missing success field is treated as installed
+        if "cert_authority_installed" in result:
+            cert_info = result["cert_authority_installed"]
+            if cert_info.get("success", True):
                 self.alertstore.medium(
-                    f"Key integrity violation detected: {result.get('key_integrity_violation')}",
+                    "Certificate authority installed: "
+                    f"{cert_info.get('subject', 'unknown')}",
                     result.get("timestamp") or "",
                     result,
                 )
-
-            # Flag certificate authority installations (potential MITM)
-            if "cert_authority_installed" in result:
-                cert_info = result["cert_authority_installed"]
+            else:
                 self.log.warning(
-                    "Certificate authority installed: %s (success: %s)",
+                    "Failed certificate authority install attempt: %s",
                     cert_info.get("subject", "unknown"),
-                    cert_info.get("success", "unknown"),
                 )
 
-            # Flag wipe failures
-            if "wipe_failure" in result:
+        # Flag wipe failures
+        if "wipe_failure" in result:
+            self.alertstore.medium(
+                "Device wipe failure detected",
+                result.get("timestamp") or "",
+                result,
+            )
+
+        # Flag crypto self test failures
+        if "crypto_self_test_completed" in result:
+            test_result = result["crypto_self_test_completed"]
+            if isinstance(test_result, dict):
+                success = test_result.get("success", True)
+            else:
+                success = test_result == 1
+            if not success:
                 self.alertstore.medium(
-                    "Device wipe failure detected",
+                    "Cryptographic self test failed",
                     result.get("timestamp") or "",
                     result,
                 )
 
-            # Flag crypto self test failures
-            if "crypto_self_test_completed" in result:
-                test_result = result["crypto_self_test_completed"]
-                if isinstance(test_result, dict):
-                    success = test_result.get("success", True)
-                else:
-                    success = test_result == 1
-                if not success:
-                    self.alertstore.medium(
-                        "Cryptographic self test failed",
-                        result.get("timestamp") or "",
-                        result,
-                    )
-
     def serialize(self, record: dict) -> Union[dict, list]:
         """Serialize a security event record for timeline output."""
         # Determine the event sub-type

src/mvt/android/parsers/backup.py

@@ -29,9 +29,6 @@ class InvalidBackupPassword(AndroidBackupParsingError):
     pass
 
 
-# TODO: Need to clean all the following code and conform it to the coding style.
-
-
 def to_utf8_bytes(input_bytes):
     output = []
     for byte in input_bytes:
@@ -51,13 +48,16 @@ def parse_ab_header(data):
         'encryption': "none", 'version': 4}
     """
     if data.startswith(b"ANDROID BACKUP"):
-        [_, version, is_compressed, encryption, _] = data.split(b"\n", 4)
-        return {
-            "backup": True,
-            "compression": (is_compressed == b"1"),
-            "version": int(version),
-            "encryption": encryption.decode("utf-8"),
-        }
+        try:
+            [_, version, is_compressed, encryption, _] = data.split(b"\n", 4)
+            return {
+                "backup": True,
+                "compression": (is_compressed == b"1"),
+                "version": int(version),
+                "encryption": encryption.decode("utf-8"),
+            }
+        except (UnicodeDecodeError, ValueError):
+            pass
 
     return {"backup": False, "compression": None, "version": None, "encryption": None}
 
@@ -157,13 +157,13 @@ def decrypt_backup_data(encrypted_backup, password, encryption_algo, format_vers
         checksum_salt=checksum_salt,
     )
 
-    # Decrypt and unpad backup data using derivied key.
+    # Decrypt and unpad backup data using derived key.
     cipher = Cipher(algorithms.AES(master_key), modes.CBC(master_iv))
     decryptor = cipher.decryptor()
     decrypted_tar = decryptor.update(encrypted_data) + decryptor.finalize()
 
     unpadder = padding.PKCS7(128).unpadder()
-    return unpadder.update(decrypted_tar)
+    return unpadder.update(decrypted_tar) + unpadder.finalize()
 
 
 def parse_backup_file(data, password=None):
@@ -210,6 +210,8 @@ def parse_tar_for_sms(data):
                 or member.name.endswith("_mms_backup")
             ):
                 dhandler = tar.extractfile(member)
+                if not dhandler:
+                    continue
                 res.extend(parse_sms_file(dhandler.read()))
 
     return res

src/mvt/common/artifact.py

@@ -8,5 +8,6 @@ from .module import MVTModule
 class Artifact(MVTModule):
     """Base class for artifacts.
 
-    XXX: Inheriting from MVTModule to have the same signature as other modules. Not sure if this is a good idea.
+    Artifacts share the MVTModule lifecycle so commands can run artifacts and
+    extraction modules through the same interface.
     """

src/mvt/common/cmd_check_iocs.py

@@ -8,6 +8,7 @@ import os
 from typing import Optional
 
 from mvt.common.command import Command
+from mvt.common.module import MVTModule
 from mvt.common.utils import exec_or_profile
 
 log = logging.getLogger(__name__)
@@ -26,6 +27,8 @@ class CmdCheckIOCS(Command):
         sub_command: Optional[bool] = False,
         disable_version_check: bool = False,
         disable_indicator_check: bool = False,
+        custom_modules: Optional[list[type[MVTModule]]] = None,
+        platform: str = "",
     ) -> None:
         super().__init__(
             target_path=target_path,
@@ -39,14 +42,16 @@ class CmdCheckIOCS(Command):
             log=log,
             disable_version_check=disable_version_check,
             disable_indicator_check=disable_indicator_check,
+            custom_modules=custom_modules,
         )
 
+        self.platform = platform
         self.name = "check-iocs"
 
     def run(self) -> None:
         assert self.target_path is not None
         all_modules = []
-        for entry in self.modules:
+        for entry in self._available_modules():
             if entry not in all_modules:
                 all_modules.append(entry)
 

src/mvt/common/command.py

@@ -8,6 +8,7 @@ import logging
 import os
 import sys
 from datetime import datetime
+from heapq import heappop, heappush
 from typing import Any, Optional
 
 from rich.console import Console
@@ -18,6 +19,8 @@ from .alerts import AlertLevel, AlertStore
 from .config import settings
 from .indicators import Indicators
 from .module import EncryptedBackupError, MVTModule, run_module, save_timeline
+from .module_loader import module_supports_command
+from .module_types import ModuleTimeline
 from .utils import (
     CustomJSONEncoder,
     convert_datetime_to_iso,
@@ -42,9 +45,12 @@ class Command:
         log: logging.Logger = logging.getLogger(__name__),
         disable_version_check: bool = False,
         disable_indicator_check: bool = False,
+        custom_modules: Optional[list[type[MVTModule]]] = None,
     ) -> None:
         self.name = ""
-        self.modules: list[Any] = []
+        self.platform = ""
+        self.modules: list[type[MVTModule]] = []
+        self.custom_modules = custom_modules if custom_modules else []
 
         self.target_path = target_path
         self.results_path = results_path
@@ -63,10 +69,10 @@ class Command:
 
         # This list will contain all executed modules.
         # We can use this to reference e.g. self.executed[0].results.
-        self.executed: list[Any] = []
+        self.executed: list[MVTModule] = []
         self.hashes = hashes
         self.hash_values: list[dict[str, Any]] = []
-        self.timeline: list[dict[str, Any]] = []
+        self.timeline: ModuleTimeline = []
 
         # Load IOCs
         self._create_storage()
@@ -197,9 +203,24 @@ class Command:
 
     def list_modules(self) -> None:
         self.log.info("Following is the list of available %s modules:", self.name)
-        for module in self.modules:
+        for module in self._available_modules():
             self.log.info(" - %s", module.__name__)
 
+    def _available_modules(self) -> list[type[MVTModule]]:
+        modules = list(self.modules)
+        modules.extend(
+            module
+            for module in self.custom_modules
+            if module_supports_command(module, self.platform, self.name)
+        )
+
+        deduplicated = []
+        for module in modules:
+            if module not in deduplicated:
+                deduplicated.append(module)
+
+        return deduplicated
+
     def init(self) -> None:
         raise NotImplementedError
 
@@ -258,22 +279,86 @@ class Command:
         console.print("")
         console.print(panel)
 
+    def _ordered_modules(self) -> Optional[list[type[MVTModule]]]:
+        """Return enabled modules in stable topological order."""
+        modules = self._available_modules()
+        module_indexes = {module: index for index, module in enumerate(modules)}
+
+        if self.module_name:
+            selected = [
+                module for module in modules if module.__name__ == self.module_name
+            ]
+        else:
+            selected = [module for module in modules if module.enabled]
+
+        required = set(selected)
+        pending = list(selected)
+        while pending:
+            module = pending.pop()
+            for dependency in module.dependencies:
+                if dependency not in module_indexes:
+                    self.log.warning(
+                        "Module %s depends on unavailable module %s. "
+                        "No modules will be run.",
+                        module.__name__,
+                        dependency.__name__,
+                    )
+                    return None
+                if dependency not in required:
+                    required.add(dependency)
+                    pending.append(dependency)
+
+        dependents: dict[type[MVTModule], list[type[MVTModule]]] = {
+            module: [] for module in required
+        }
+        indegree = {module: 0 for module in required}
+        for module in required:
+            for dependency in module.dependencies:
+                if dependency not in required:
+                    continue
+                dependents[dependency].append(module)
+                indegree[module] += 1
+
+        ready: list[tuple[int, type[MVTModule]]] = []
+        for module, count in indegree.items():
+            if count == 0:
+                heappush(ready, (module_indexes[module], module))
+
+        ordered = []
+        while ready:
+            _, module = heappop(ready)
+            ordered.append(module)
+            for dependent in dependents[module]:
+                indegree[dependent] -= 1
+                if indegree[dependent] == 0:
+                    heappush(ready, (module_indexes[dependent], dependent))
+
+        if len(ordered) != len(required):
+            cyclic_modules = sorted(
+                (module.__name__ for module, count in indegree.items() if count > 0)
+            )
+            self.log.warning(
+                "Circular module dependency detected involving: %s. "
+                "No modules will be run.",
+                ", ".join(cyclic_modules),
+            )
+            return None
+
+        return ordered
+
     def run(self) -> None:
+        ordered_modules = self._ordered_modules()
+        if ordered_modules is None:
+            return
+
         try:
             self.init()
         except NotImplementedError:
             pass
 
-        for module in self.modules:
-            if self.module_name and module.__name__ != self.module_name:
-                continue
+        executed_by_type: dict[type[MVTModule], MVTModule] = {}
+        for module in ordered_modules:
 
-            if not module.enabled and not (
-                self.module_name and module.__name__ == self.module_name
-            ):
-                continue
-
-            # FIXME: do we need the logger here
             module_logger = logging.getLogger(module.__module__)
 
             m = module(
@@ -282,6 +367,10 @@ class Command:
                 module_options=self.module_options,
                 log=module_logger,
             )
+            m.dependency_modules = {
+                dependency: executed_by_type[dependency]
+                for dependency in module.dependencies
+            }
 
             if self.iocs.total_ioc_count:
                 m.indicators = self.iocs
@@ -305,6 +394,7 @@ class Command:
                 return
 
             self.executed.append(m)
+            executed_by_type[module] = m
             self.timeline.extend(m.timeline)
             self.alertstore.extend(m.alertstore.alerts)
 

src/mvt/common/completion.py

@@ -0,0 +1,94 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2023 The MVT Authors.
+# Use of this software is governed by the MVT License 1.1 that can be found at
+#   https://license.mvt.re/1.1/
+
+from pathlib import Path
+import shlex
+
+import click
+from click.shell_completion import get_completion_class
+
+
+SUPPORTED_SHELLS = ("bash", "zsh", "fish")
+
+
+def completion_instructions(program_name: str) -> str:
+    return f"""Shell completion for {program_name}
+
+Print a completion script:
+  {program_name} completion bash > ~/.{program_name}-complete.bash
+  {program_name} completion zsh > ~/.{program_name}-complete.zsh
+  mkdir -p ~/.config/fish/completions
+  {program_name} completion fish > ~/.config/fish/completions/{program_name}.fish
+
+Load the generated Bash script from ~/.bashrc:
+  [ -f ~/.{program_name}-complete.bash ] && . ~/.{program_name}-complete.bash
+
+Load the generated Zsh script from ~/.zshrc:
+  [ -f ~/.{program_name}-complete.zsh ] && . ~/.{program_name}-complete.zsh
+
+Fish loads completion files from ~/.config/fish/completions automatically.
+
+To write these files and update Bash/Zsh shell configuration automatically:
+  {program_name} completion bash --install
+  {program_name} completion zsh --install
+  {program_name} completion fish --install
+"""
+
+
+def generate_completion_script(cli: click.Command, program_name: str, shell: str) -> str:
+    completion_class = get_completion_class(shell)
+    if completion_class is None:
+        raise click.ClickException(f"Unsupported shell: {shell}")
+
+    complete_var = f"_{program_name.upper().replace('-', '_')}_COMPLETE"
+    return completion_class(cli, {}, program_name, complete_var).source()
+
+
+def install_completion_script(
+    cli: click.Command,
+    program_name: str,
+    shell: str,
+) -> Path:
+    script = generate_completion_script(cli, program_name, shell)
+    script_path = _completion_script_path(program_name, shell)
+    script_path.parent.mkdir(parents=True, exist_ok=True)
+    script_path.write_text(script, encoding="utf-8")
+
+    if shell in ("bash", "zsh"):
+        _install_shell_source_line(program_name, shell, script_path)
+
+    return script_path
+
+
+def _completion_script_path(program_name: str, shell: str) -> Path:
+    home = Path.home()
+
+    if shell == "fish":
+        return home / ".config" / "fish" / "completions" / f"{program_name}.fish"
+
+    return home / f".{program_name}-complete.{shell}"
+
+
+def _install_shell_source_line(program_name: str, shell: str, script_path: Path) -> None:
+    shell_config_path = Path.home() / f".{shell}rc"
+    source_line = (
+        f"[ -f {shlex.quote(str(script_path))} ] && "
+        f". {shlex.quote(str(script_path))}"
+    )
+    block = (
+        f"# MVT shell completion for {program_name}\n"
+        f"{source_line}\n"
+    )
+
+    if shell_config_path.exists():
+        shell_config = shell_config_path.read_text(encoding="utf-8")
+        if source_line in shell_config:
+            return
+    else:
+        shell_config = ""
+
+    separator = "" if not shell_config or shell_config.endswith("\n") else "\n"
+    with shell_config_path.open("a", encoding="utf-8") as handle:
+        handle.write(f"{separator}{block}")

src/mvt/common/help.py

@@ -10,6 +10,10 @@ HELP_MSG_IOC = "Path to indicators file (can be invoked multiple time)"
 HELP_MSG_FAST = "Avoid running time/resource consuming features"
 HELP_MSG_LIST_MODULES = "Print list of available modules and exit"
 HELP_MSG_MODULE = "Name of a single module you would like to run instead of all"
+HELP_MSG_LOAD_MODULE = (
+    "Load custom MVT module(s) from a Python file or folder "
+    "(can be invoked multiple times)"
+)
 HELP_MSG_NONINTERACTIVE = "Don't ask interactive questions during processing"
 HELP_MSG_HASHES = "Generate hashes of all the files analyzed"
 HELP_MSG_VERBOSE = "Verbose mode"
@@ -17,6 +21,7 @@ HELP_MSG_CHECK_IOCS = "Compare stored JSON results to provided indicators"
 HELP_MSG_STIX2 = "Download public STIX2 indicators"
 HELP_MSG_DISABLE_UPDATE_CHECK = "Disable MVT version update check"
 HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK = "Disable indicators update check"
+HELP_MSG_COMPLETION = "Generate or install shell completion"
 
 # IOS Specific
 HELP_MSG_DECRYPT_BACKUP = "Decrypt an encrypted iTunes backup"

src/mvt/common/module.py

@@ -9,7 +9,7 @@ import logging
 import os
 import re
 from dataclasses import asdict, is_dataclass
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Optional, Sequence
 
 from .alerts import AlertStore
 from .indicators import Indicators
@@ -43,6 +43,8 @@ class MVTModule:
 
     enabled: bool = True
     slug: Optional[str] = None
+    dependencies: Sequence[type["MVTModule"]] = ()
+    supported_commands: Sequence[tuple[str, str]] = ()
 
     def __init__(
         self,
@@ -51,7 +53,7 @@ class MVTModule:
         results_path: Optional[str] = None,
         module_options: Optional[Dict[str, Any]] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         """Initialize module.
 
@@ -71,16 +73,22 @@ class MVTModule:
         self.file_path: Optional[str] = file_path
         self.target_path: Optional[str] = target_path
         self.results_path: Optional[str] = results_path
-        self.module_options: Optional[Dict[str, Any]] = (
-            module_options if module_options else {}
-        )
+        self.serial: Optional[str] = None
+        self.module_options: Dict[str, Any] = module_options if module_options else {}
 
         self.log = log
         self.indicators: Optional[Indicators] = None
         self.alertstore: AlertStore = AlertStore(log=log)
 
-        self.results: ModuleResults = results if results else []
+        self.results: ModuleResults = results if results is not None else []
         self.timeline: ModuleTimeline = []
+        self.dependency_modules: Dict[type["MVTModule"], "MVTModule"] = {}
+
+    def get_dependency_results(
+        self, module_class: type["MVTModule"]
+    ) -> ModuleResults:
+        """Return the results produced by a prerequisite module."""
+        return self.dependency_modules[module_class].results
 
     @classmethod
     def from_json(cls, json_path: str, log: logging.Logger):
@@ -109,11 +117,14 @@ class MVTModule:
         name = self.get_slug()
 
         if self.results:
+            converted_results: Any
             if isinstance(self.results, dict):
                 converted_results = self.results
             else:
                 converted_results = [
-                    asdict(result) if is_dataclass(result) else result
+                    asdict(result)
+                    if is_dataclass(result) and not isinstance(result, type)
+                    else result
                     for result in self.results
                 ]
             results_file_name = f"{name}.json"

src/mvt/common/module_loader.py

@@ -0,0 +1,127 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2026 The MVT Authors.
+# Use of this software is governed by the MVT License 1.1 that can be found at
+#   https://license.mvt.re/1.1/
+
+import hashlib
+import importlib.util
+import inspect
+import os
+import sys
+from pathlib import Path
+from types import ModuleType
+from typing import Iterable, Optional
+
+from .module import MVTModule
+
+MVT_CUSTOM_MODULES_ENV = "MVT_CUSTOM_MODULES"
+
+
+class CustomModuleLoadError(Exception):
+    pass
+
+
+def _module_name_for_path(path: Path) -> str:
+    digest = hashlib.sha256(str(path).encode("utf-8")).hexdigest()[:16]
+    return f"_mvt_custom_module_{path.stem}_{digest}"
+
+
+def _iter_module_files(path: Path) -> Iterable[Path]:
+    if path.is_file():
+        if path.suffix != ".py":
+            raise CustomModuleLoadError(f"Custom module file is not a Python file: {path}")
+        yield path
+        return
+
+    if path.is_dir():
+        for child in sorted(path.iterdir()):
+            if child.name.startswith("."):
+                continue
+            if child.name == "__init__.py":
+                continue
+            if child.is_file() and child.suffix == ".py":
+                yield child
+        return
+
+    raise CustomModuleLoadError(f"Custom module path does not exist: {path}")
+
+
+def _load_python_file(path: Path) -> ModuleType:
+    module_name = _module_name_for_path(path)
+    spec = importlib.util.spec_from_file_location(module_name, path)
+    if spec is None or spec.loader is None:
+        raise CustomModuleLoadError(f"Unable to load custom module file: {path}")
+
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[module_name] = module
+    try:
+        spec.loader.exec_module(module)
+    except Exception as exc:
+        raise CustomModuleLoadError(f"Unable to import custom module {path}: {exc}") from exc
+
+    return module
+
+
+def discover_mvt_modules(module: ModuleType) -> list[type[MVTModule]]:
+    modules = []
+    for _, obj in inspect.getmembers(module, inspect.isclass):
+        if obj is MVTModule:
+            continue
+        if obj.__module__ != module.__name__:
+            continue
+        if not issubclass(obj, MVTModule):
+            continue
+        modules.append(obj)
+
+    return modules
+
+
+def load_custom_modules_from_path(path: str) -> list[type[MVTModule]]:
+    custom_modules: list[type[MVTModule]] = []
+    seen: set[tuple[str, str]] = set()
+    resolved_path = Path(path).expanduser().resolve()
+
+    for module_file in _iter_module_files(resolved_path):
+        loaded_module = _load_python_file(module_file)
+        for module_class in discover_mvt_modules(loaded_module):
+            key = (str(module_file), module_class.__qualname__)
+            if key in seen:
+                continue
+            seen.add(key)
+            custom_modules.append(module_class)
+
+    return custom_modules
+
+
+def load_custom_modules(paths: Optional[Iterable[str]] = None) -> list[type[MVTModule]]:
+    search_paths: list[str] = []
+    env_path = os.environ.get(MVT_CUSTOM_MODULES_ENV)
+    if env_path:
+        search_paths.append(env_path)
+    if paths:
+        search_paths.extend(paths)
+
+    custom_modules: list[type[MVTModule]] = []
+    seen: set[tuple[str, str]] = set()
+    for path in search_paths:
+        for module_class in load_custom_modules_from_path(path):
+            source = Path(inspect.getfile(module_class)).resolve()
+            key = (str(source), module_class.__qualname__)
+            if key in seen:
+                continue
+            seen.add(key)
+            custom_modules.append(module_class)
+
+    return custom_modules
+
+
+def module_supports_command(
+    module_class: type[MVTModule],
+    platform: str,
+    command: str,
+) -> bool:
+    supported_commands = getattr(module_class, "supported_commands", None)
+    if not supported_commands:
+        return True
+
+    return (platform, command) in {tuple(entry) for entry in supported_commands}

src/mvt/common/module_types.py

@@ -16,7 +16,10 @@ from typing import Any, Dict, List, Union
 ModuleAtomicResult = Dict[str, Any]
 
 
-ModuleResults = List[ModuleAtomicResult]
+# Extraction modules historically use either a list of records or grouped
+# dictionaries keyed by source path. Keep this alias broad until those shapes
+# are modeled per module.
+ModuleResults = Any
 
 
 @dataclass

src/mvt/common/updates.py

@@ -180,10 +180,8 @@ class IndicatorsUpdates:
     def _get_remote_file_latest_commit(
         self, owner: str, repo: str, branch: str, path: str
     ) -> int:
-        # TODO: The branch is currently not taken into consideration.
-        #       How do we specify which branch to look up to the API?
         file_commit_url = (
-            f"https://api.github.com/repos/{owner}/{repo}/commits?path={path}"
+            f"https://api.github.com/repos/{owner}/{repo}/commits?path={path}&sha={branch}"
         )
         try:
             res = requests.get(file_commit_url, timeout=5)

src/mvt/common/url.py

@@ -3,11 +3,16 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
+import logging
 from typing import Optional
 
 import requests
 from tld import get_tld
 
+from .config import settings
+
+log = logging.getLogger(__name__)
+
 SHORTENER_DOMAINS = [
     "0rz.tw",
     "1drv.ms",
@@ -375,7 +380,16 @@ class URL:
 
     def unshorten(self) -> Optional[str]:
         """Unshorten the URL by requesting an HTTP HEAD response."""
-        res = requests.head(self.url)
+
+        if settings.NETWORK_ACCESS_ALLOWED is False:
+            log.info(
+                "Network access disabled (MVT_NETWORK_ACCESS_ALLOWED=False), "
+                "skipping unshorten for %s",
+                self.url,
+            )
+            return ""
+
+        res = requests.head(self.url, timeout=settings.NETWORK_TIMEOUT)
         if str(res.status_code).startswith("30"):
             return res.headers["Location"]
 

src/mvt/common/utils.py

@@ -123,10 +123,9 @@ def convert_mactime_to_datetime(timestamp: Union[int, float], from_2001: bool =
     if from_2001:
         timestamp = timestamp + 978307200
 
-    # TODO: This is rather ugly. Happens sometimes with invalid timestamps.
     try:
         return convert_unix_to_utc_datetime(timestamp)
-    except Exception:
+    except (OSError, OverflowError, ValueError):
         return None
 
 

src/mvt/ios/cli.py

@@ -11,6 +11,12 @@ import click
 from rich.prompt import Prompt
 
 from mvt.common.cmd_check_iocs import CmdCheckIOCS
+from mvt.common.completion import (
+    SUPPORTED_SHELLS,
+    completion_instructions,
+    generate_completion_script,
+    install_completion_script,
+)
 from mvt.common.logo import logo
 from mvt.common.options import MutuallyExclusiveOption
 from mvt.common.updates import IndicatorsUpdates
@@ -31,6 +37,7 @@ from mvt.common.help import (
     HELP_MSG_OUTPUT,
     HELP_MSG_FAST,
     HELP_MSG_LIST_MODULES,
+    HELP_MSG_LOAD_MODULE,
     HELP_MSG_MODULE,
     HELP_MSG_VERBOSE,
     HELP_MSG_CHECK_FS,
@@ -39,7 +46,9 @@ from mvt.common.help import (
     HELP_MSG_CHECK_IOS_BACKUP,
     HELP_MSG_DISABLE_UPDATE_CHECK,
     HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK,
+    HELP_MSG_COMPLETION,
 )
+from mvt.common.module_loader import CustomModuleLoadError, load_custom_modules
 from .cmd_check_backup import CmdIOSCheckBackup
 from .cmd_check_fs import CmdIOSCheckFS
 from .decrypt import DecryptBackup
@@ -65,6 +74,13 @@ def _get_disable_flags(ctx):
     )
 
 
+def _load_custom_modules(load_module):
+    try:
+        return load_custom_modules(load_module)
+    except CustomModuleLoadError as exc:
+        raise click.ClickException(str(exc)) from exc
+
+
 # ==============================================================================
 # Main
 # ==============================================================================
@@ -82,10 +98,11 @@ def cli(ctx, disable_update_check, disable_indicator_update_check):
     ctx.ensure_object(dict)
     ctx.obj["disable_version_check"] = disable_update_check
     ctx.obj["disable_indicator_check"] = disable_indicator_update_check
-    logo(
-        disable_version_check=disable_update_check,
-        disable_indicator_check=disable_indicator_update_check,
-    )
+    if ctx.invoked_subcommand != "completion":
+        logo(
+            disable_version_check=disable_update_check,
+            disable_indicator_check=disable_indicator_update_check,
+        )
 
 
 # ==============================================================================
@@ -96,6 +113,40 @@ def version():
     return
 
 
+# ==============================================================================
+# Command: completion
+# ==============================================================================
+@cli.command("completion", context_settings=CONTEXT_SETTINGS, help=HELP_MSG_COMPLETION)
+@click.argument("shell", required=False, type=click.Choice(SUPPORTED_SHELLS))
+@click.option(
+    "--install",
+    is_flag=True,
+    help="Write completion files and update shell configuration.",
+)
+@click.pass_context
+def completion(ctx, shell, install):
+    program_name = "mvt-ios"
+
+    if shell is None:
+        if install:
+            raise click.UsageError("A shell is required when using --install.")
+        click.echo(completion_instructions(program_name))
+        return
+
+    root_cli = ctx.find_root().command
+
+    if install:
+        script_path = install_completion_script(root_cli, program_name, shell)
+        click.echo(f"Installed {shell} completion to {script_path}")
+        if shell in ("bash", "zsh"):
+            click.echo(f"Updated ~/.{shell}rc")
+        else:
+            click.echo("Fish loads completion files automatically.")
+        return
+
+    click.echo(generate_completion_script(root_cli, program_name, shell))
+
+
 # ==============================================================================
 # Command: decrypt-backup
 # ==============================================================================
@@ -229,15 +280,32 @@ def extract_key(password, key_file, backup_path):
 @click.option("--fast", "-f", is_flag=True, help=HELP_MSG_FAST)
 @click.option("--list-modules", "-l", is_flag=True, help=HELP_MSG_LIST_MODULES)
 @click.option("--module", "-m", help=HELP_MSG_MODULE)
+@click.option(
+    "--load-module",
+    type=click.Path(exists=True),
+    multiple=True,
+    default=[],
+    help=HELP_MSG_LOAD_MODULE,
+)
 @click.option("--hashes", "-H", is_flag=True, help=HELP_MSG_HASHES)
 @click.option("--verbose", "-v", is_flag=True, help=HELP_MSG_VERBOSE)
 @click.argument("BACKUP_PATH", type=click.Path(exists=True))
 @click.pass_context
 def check_backup(
-    ctx, iocs, output, fast, list_modules, module, hashes, verbose, backup_path
+    ctx,
+    iocs,
+    output,
+    fast,
+    list_modules,
+    module,
+    load_module,
+    hashes,
+    verbose,
+    backup_path,
 ):
     set_verbose_logging(verbose)
     module_options = {"fast_mode": fast}
+    custom_modules = _load_custom_modules(load_module)
 
     cmd = CmdIOSCheckBackup(
         target_path=backup_path,
@@ -248,6 +316,7 @@ def check_backup(
         hashes=hashes,
         disable_version_check=_get_disable_flags(ctx)[0],
         disable_indicator_check=_get_disable_flags(ctx)[1],
+        custom_modules=custom_modules,
     )
 
     if list_modules:
@@ -277,13 +346,32 @@ def check_backup(
 @click.option("--fast", "-f", is_flag=True, help=HELP_MSG_FAST)
 @click.option("--list-modules", "-l", is_flag=True, help=HELP_MSG_LIST_MODULES)
 @click.option("--module", "-m", help=HELP_MSG_MODULE)
+@click.option(
+    "--load-module",
+    type=click.Path(exists=True),
+    multiple=True,
+    default=[],
+    help=HELP_MSG_LOAD_MODULE,
+)
 @click.option("--hashes", "-H", is_flag=True, help=HELP_MSG_HASHES)
 @click.option("--verbose", "-v", is_flag=True, help=HELP_MSG_VERBOSE)
 @click.argument("DUMP_PATH", type=click.Path(exists=True))
 @click.pass_context
-def check_fs(ctx, iocs, output, fast, list_modules, module, hashes, verbose, dump_path):
+def check_fs(
+    ctx,
+    iocs,
+    output,
+    fast,
+    list_modules,
+    module,
+    load_module,
+    hashes,
+    verbose,
+    dump_path,
+):
     set_verbose_logging(verbose)
     module_options = {"fast_mode": fast}
+    custom_modules = _load_custom_modules(load_module)
 
     cmd = CmdIOSCheckFS(
         target_path=dump_path,
@@ -294,6 +382,7 @@ def check_fs(ctx, iocs, output, fast, list_modules, module, hashes, verbose, dum
         hashes=hashes,
         disable_version_check=_get_disable_flags(ctx)[0],
         disable_indicator_check=_get_disable_flags(ctx)[1],
+        custom_modules=custom_modules,
     )
 
     if list_modules:
@@ -321,15 +410,25 @@ def check_fs(ctx, iocs, output, fast, list_modules, module, hashes, verbose, dum
 )
 @click.option("--list-modules", "-l", is_flag=True, help=HELP_MSG_LIST_MODULES)
 @click.option("--module", "-m", help=HELP_MSG_MODULE)
+@click.option(
+    "--load-module",
+    type=click.Path(exists=True),
+    multiple=True,
+    default=[],
+    help=HELP_MSG_LOAD_MODULE,
+)
 @click.argument("FOLDER", type=click.Path(exists=True))
 @click.pass_context
-def check_iocs(ctx, iocs, list_modules, module, folder):
+def check_iocs(ctx, iocs, list_modules, module, load_module, folder):
+    custom_modules = _load_custom_modules(load_module)
     cmd = CmdCheckIOCS(
         target_path=folder,
         ioc_files=iocs,
         module_name=module,
         disable_version_check=_get_disable_flags(ctx)[0],
         disable_indicator_check=_get_disable_flags(ctx)[1],
+        custom_modules=custom_modules,
+        platform="ios",
     )
     cmd.modules = BACKUP_MODULES + FS_MODULES + MIXED_MODULES
 

src/mvt/ios/cmd_check_backup.py

@@ -8,6 +8,7 @@ from typing import Optional
 
 from mvt.common.command import Command
 from mvt.common.indicators import Indicators
+from mvt.common.module import MVTModule
 
 from .modules.backup import BACKUP_MODULES
 from .modules.mixed import MIXED_MODULES
@@ -29,6 +30,7 @@ class CmdIOSCheckBackup(Command):
         sub_command: bool = False,
         disable_version_check: bool = False,
         disable_indicator_check: bool = False,
+        custom_modules: Optional[list[type[MVTModule]]] = None,
     ) -> None:
         super().__init__(
             target_path=target_path,
@@ -43,8 +45,10 @@ class CmdIOSCheckBackup(Command):
             log=log,
             disable_version_check=disable_version_check,
             disable_indicator_check=disable_indicator_check,
+            custom_modules=custom_modules,
         )
 
+        self.platform = "ios"
         self.name = "check-backup"
         self.modules = BACKUP_MODULES + MIXED_MODULES
 

src/mvt/ios/cmd_check_fs.py

@@ -8,6 +8,7 @@ from typing import Optional
 
 from mvt.common.command import Command
 from mvt.common.indicators import Indicators
+from mvt.common.module import MVTModule
 
 from .modules.fs import FS_MODULES
 from .modules.mixed import MIXED_MODULES
@@ -29,6 +30,7 @@ class CmdIOSCheckFS(Command):
         sub_command: bool = False,
         disable_version_check: bool = False,
         disable_indicator_check: bool = False,
+        custom_modules: Optional[list[type[MVTModule]]] = None,
     ) -> None:
         super().__init__(
             target_path=target_path,
@@ -42,8 +44,10 @@ class CmdIOSCheckFS(Command):
             log=log,
             disable_version_check=disable_version_check,
             disable_indicator_check=disable_indicator_check,
+            custom_modules=custom_modules,
         )
 
+        self.platform = "ios"
         self.name = "check-fs"
         self.modules = FS_MODULES + MIXED_MODULES
 

src/mvt/ios/modules/backup/backup_info.py

@@ -25,7 +25,7 @@ class BackupInfo(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,
@@ -36,7 +36,7 @@ class BackupInfo(IOSExtraction):
             results=results,
         )
 
-        self.results: dict = {}
+        self.results: dict = results if results is not None else {}
 
     def run(self) -> None:
         if not self.target_path:

src/mvt/ios/modules/backup/configuration_profiles.py

@@ -33,7 +33,7 @@ class ConfigurationProfiles(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,
@@ -85,6 +85,35 @@ class ConfigurationProfiles(IOSExtraction):
                     self.alertstore.medium(warning_message, "", result)
                     continue
 
+    @staticmethod
+    def _b64encode_key(d: dict, key: str) -> None:
+        if key in d:
+            d[key] = b64encode(d[key])
+
+    @staticmethod
+    def _b64encode_keys(d: dict, keys: list) -> None:
+        for key in keys:
+            if key in d:
+                d[key] = b64encode(d[key])
+
+    def _b64encode_plist_bytes(self, plist: dict) -> None:
+        """Encode binary plist values to base64 for JSON serialization."""
+        if "SignerCerts" in plist:
+            plist["SignerCerts"] = [b64encode(x) for x in plist["SignerCerts"]]
+
+        self._b64encode_keys(plist, ["PushTokenDataSentToServerKey", "LastPushTokenHash"])
+
+        if "OTAProfileStub" in plist:
+            stub = plist["OTAProfileStub"]
+            if "SignerCerts" in stub:
+                stub["SignerCerts"] = [b64encode(x) for x in stub["SignerCerts"]]
+            if "PayloadContent" in stub:
+                self._b64encode_key(stub["PayloadContent"], "EnrollmentIdentityPersistentID")
+
+        if "PayloadContent" in plist:
+            for entry in plist["PayloadContent"]:
+                self._b64encode_keys(entry, ["PERSISTENT_REF", "IdentityPersistentRef"])
+
     def run(self) -> None:
         for conf_file in self._get_backup_files_from_manifest(
             domain=CONF_PROFILES_DOMAIN
@@ -113,65 +142,7 @@ class ConfigurationProfiles(IOSExtraction):
                 except Exception:
                     conf_plist = {}
 
-            # TODO: Tidy up the following code hell.
-
-            if "SignerCerts" in conf_plist:
-                conf_plist["SignerCerts"] = [
-                    b64encode(x) for x in conf_plist["SignerCerts"]
-                ]
-
-            if "OTAProfileStub" in conf_plist:
-                if "SignerCerts" in conf_plist["OTAProfileStub"]:
-                    conf_plist["OTAProfileStub"]["SignerCerts"] = [
-                        b64encode(x)
-                        for x in conf_plist["OTAProfileStub"]["SignerCerts"]
-                    ]
-
-                if "PayloadContent" in conf_plist["OTAProfileStub"]:
-                    if (
-                        "EnrollmentIdentityPersistentID"
-                        in conf_plist["OTAProfileStub"]["PayloadContent"]
-                    ):
-                        conf_plist["OTAProfileStub"]["PayloadContent"][
-                            "EnrollmentIdentityPersistentID"
-                        ] = b64encode(
-                            conf_plist["OTAProfileStub"]["PayloadContent"][
-                                "EnrollmentIdentityPersistentID"
-                            ]
-                        )
-
-            if "PushTokenDataSentToServerKey" in conf_plist:
-                conf_plist["PushTokenDataSentToServerKey"] = b64encode(
-                    conf_plist["PushTokenDataSentToServerKey"]
-                )
-
-            if "LastPushTokenHash" in conf_plist:
-                conf_plist["LastPushTokenHash"] = b64encode(
-                    conf_plist["LastPushTokenHash"]
-                )
-
-            if "PayloadContent" in conf_plist:
-                for content_entry in range(len(conf_plist["PayloadContent"])):
-                    if "PERSISTENT_REF" in conf_plist["PayloadContent"][content_entry]:
-                        conf_plist["PayloadContent"][content_entry][
-                            "PERSISTENT_REF"
-                        ] = b64encode(
-                            conf_plist["PayloadContent"][content_entry][
-                                "PERSISTENT_REF"
-                            ]
-                        )
-
-                    if (
-                        "IdentityPersistentRef"
-                        in conf_plist["PayloadContent"][content_entry]
-                    ):
-                        conf_plist["PayloadContent"][content_entry][
-                            "IdentityPersistentRef"
-                        ] = b64encode(
-                            conf_plist["PayloadContent"][content_entry][
-                                "IdentityPersistentRef"
-                            ]
-                        )
+            self._b64encode_plist_bytes(conf_plist)
 
             self.results.append(
                 {

src/mvt/ios/modules/backup/manifest.py

@@ -33,7 +33,7 @@ class Manifest(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/ios/modules/backup/profile_events.py

@@ -34,7 +34,7 @@ class ProfileEvents(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/ios/modules/base.py

@@ -31,7 +31,7 @@ class IOSExtraction(MVTModule):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,
@@ -53,7 +53,7 @@ class IOSExtraction(MVTModule):
         :param file_path: Path to the malformed database file.
 
         """
-        # TODO: Find a better solution.
+        # SQLite's immutable mode cannot open databases with active WAL files.
         if not forced:
             # If the database is open, do not use immutable
             if os.path.isfile(file_path + "-shm"):

src/mvt/ios/modules/fs/analytics.py

@@ -34,7 +34,7 @@ class Analytics(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,
@@ -44,7 +44,7 @@ class Analytics(IOSExtraction):
             log=log,
             results=results,
         )
-        self.results: list = []
+        self.results: list = [] if results is None else results
 
     def serialize(self, record: ModuleAtomicResult) -> ModuleSerializedResult:
         return {

src/mvt/ios/modules/fs/analytics_ios_versions.py

@@ -30,7 +30,7 @@ class AnalyticsIOSVersions(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/ios/modules/fs/cache_files.py

@@ -25,7 +25,7 @@ class CacheFiles(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,
@@ -99,6 +99,10 @@ class CacheFiles(IOSExtraction):
 
     def run(self) -> None:
         self.results: dict = {}
+        if not self.target_path:
+            self.log.error("No filesystem dump path provided")
+            return
+
         for root, _, files in os.walk(self.target_path):
             for file_name in files:
                 if file_name != "Cache.db":

src/mvt/ios/modules/fs/filesystem.py

@@ -29,7 +29,7 @@ class Filesystem(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,
@@ -73,6 +73,10 @@ class Filesystem(IOSExtraction):
                 )
 
     def run(self) -> None:
+        if not self.target_path:
+            self.log.error("No filesystem dump path provided")
+            return
+
         for root, dirs, files in os.walk(self.target_path):
             for dir_name in dirs:
                 try:

src/mvt/ios/modules/fs/net_netusage.py

@@ -30,7 +30,7 @@ class Netusage(NetBase):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/ios/modules/fs/safari_favicon.py

@@ -31,7 +31,7 @@ class SafariFavicon(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,
@@ -41,7 +41,7 @@ class SafariFavicon(IOSExtraction):
             log=log,
             results=results,
         )
-        self.results: list = []
+        self.results: list = [] if results is None else results
 
     def serialize(self, record: ModuleAtomicResult) -> ModuleSerializedResult:
         return {

src/mvt/ios/modules/fs/shutdownlog.py

@@ -30,7 +30,7 @@ class ShutdownLog(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,
@@ -78,7 +78,7 @@ class ShutdownLog(IOSExtraction):
         recent_processes = []
         times_delayed = 0
         delay = 0.0
-        for line in content.split("\n"):
+        for line in content.splitlines():
             line = line.strip()
 
             if line.startswith("remaining client pid:"):

src/mvt/ios/modules/fs/version_history.py

@@ -32,7 +32,7 @@ class IOSVersionHistory(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,
@@ -42,7 +42,7 @@ class IOSVersionHistory(IOSExtraction):
             log=log,
             results=results,
         )
-        self.results: list = []
+        self.results: list = [] if results is None else results
 
     def serialize(self, record: ModuleAtomicResult) -> ModuleSerializedResult:
         return {

src/mvt/ios/modules/fs/webkit_indexeddb.py

@@ -34,7 +34,7 @@ class WebkitIndexedDB(WebkitBase):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/ios/modules/fs/webkit_localstorage.py

@@ -32,7 +32,7 @@ class WebkitLocalStorage(WebkitBase):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/ios/modules/fs/webkit_safariviewservice.py

@@ -28,7 +28,7 @@ class WebkitSafariViewService(WebkitBase):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/ios/modules/mixed/applications.py

@@ -41,7 +41,7 @@ class Applications(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/ios/modules/mixed/calendar.py

@@ -31,7 +31,7 @@ class Calendar(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/ios/modules/mixed/calls.py

@@ -27,7 +27,7 @@ class Calls(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: list = [],
+        results: Optional[list] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/ios/modules/mixed/chrome_favicon.py

@@ -16,7 +16,6 @@ from mvt.common.utils import convert_chrometime_to_datetime, convert_datetime_to
 from ..base import IOSExtraction
 
 CHROME_FAVICON_BACKUP_IDS = ["55680ab883d0fdcffd94f959b1632e5fbbb18c5b"]
-# TODO: Confirm Chrome database path.
 CHROME_FAVICON_ROOT_PATHS = [
     "private/var/mobile/Containers/Data/Application/*/Library/Application Support/Google/Chrome/Default/Favicons",
 ]
@@ -32,7 +31,7 @@ class ChromeFavicon(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/ios/modules/mixed/chrome_history.py

@@ -18,7 +18,6 @@ from ..base import IOSExtraction
 CHROME_HISTORY_BACKUP_IDS = [
     "faf971ce92c3ac508c018dce1bef2a8b8e9838f1",
 ]
-# TODO: Confirm Chrome database path.
 CHROME_HISTORY_ROOT_PATHS = [
     "private/var/mobile/Containers/Data/Application/*/Library/Application Support/Google/Chrome/Default/History",  # pylint: disable=line-too-long
 ]
@@ -34,7 +33,7 @@ class ChromeHistory(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/ios/modules/mixed/contacts.py

@@ -29,7 +29,7 @@ class Contacts(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/ios/modules/mixed/firefox_favicon.py

@@ -33,7 +33,7 @@ class FirefoxFavicon(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/ios/modules/mixed/firefox_history.py

@@ -37,7 +37,7 @@ class FirefoxHistory(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/ios/modules/mixed/global_preferences.py

@@ -27,7 +27,7 @@ class GlobalPreferences(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/ios/modules/mixed/idstatuscache.py

@@ -36,7 +36,7 @@ class IDStatusCache(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/ios/modules/mixed/interactionc.py

@@ -228,7 +228,7 @@ class InteractionC(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/ios/modules/mixed/locationd.py

@@ -36,7 +36,7 @@ class LocationdClients(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,
@@ -147,7 +147,6 @@ class LocationdClients(IOSExtraction):
             # Some migration information are int and not dicts
             if not isinstance(file_plist[key], dict):
                 continue
-            # FIXME: unclear key format in iOS 17
             result = file_plist[key]
             result["package"] = key.rstrip(":")
             for timestamp in self.timestamps:

src/mvt/ios/modules/mixed/net_datausage.py

@@ -31,7 +31,7 @@ class Datausage(NetBase):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/ios/modules/mixed/osanalytics_addaily.py

@@ -35,7 +35,7 @@ class OSAnalyticsADDaily(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/ios/modules/mixed/safari_browserstate.py

@@ -36,7 +36,7 @@ class SafariBrowserState(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/ios/modules/mixed/safari_history.py

@@ -38,7 +38,7 @@ class SafariHistory(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/ios/modules/mixed/shortcuts.py

@@ -37,7 +37,7 @@ class Shortcuts(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/ios/modules/mixed/sms.py

@@ -35,7 +35,7 @@ class SMS(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/ios/modules/mixed/sms_attachments.py

@@ -35,7 +35,7 @@ class SMSAttachments(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/ios/modules/mixed/tcc.py

@@ -56,7 +56,7 @@ class TCC(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/ios/modules/mixed/webkit_resource_load_statistics.py

@@ -37,7 +37,7 @@ class WebkitResourceLoadStatistics(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,
@@ -85,32 +85,56 @@ class WebkitResourceLoadStatistics(IOSExtraction):
 
         try:
             try:
-                # FIXME: table contains extra fields with timestamp here
+                cur.execute("PRAGMA table_info(ObservedDomains);")
+                available_columns = {row[1] for row in cur}
+                required_columns = [
+                    "domainID",
+                    "registrableDomain",
+                    "lastSeen",
+                    "hadUserInteraction",
+                ]
+                if not set(required_columns).issubset(available_columns):
+                    return
+
+                optional_columns = [
+                    column
+                    for column in [
+                        "mostRecentUserInteractionTime",
+                        "mostRecentWebPushInteractionTime",
+                    ]
+                    if column in available_columns
+                ]
+                selected_columns = required_columns + optional_columns
                 cur.execute(
-                    """
-                    SELECT
-                        domainID,
-                        registrableDomain,
-                        lastSeen,
-                        hadUserInteraction
-                    from ObservedDomains;
-                """
+                    f"SELECT {', '.join(selected_columns)} FROM ObservedDomains;"
                 )
             except sqlite3.OperationalError:
                 return
 
             for row in cur:
-                self.results.append(
-                    {
-                        "domain_id": row[0],
-                        "registrable_domain": row[1],
-                        "last_seen": row[2],
-                        "had_user_interaction": bool(row[3]),
-                        "last_seen_isodate": convert_unix_to_iso(row[2]),
-                        "domain": domain,
-                        "path": path,
-                    }
-                )
+                result = {
+                    "domain_id": row[0],
+                    "registrable_domain": row[1],
+                    "last_seen": row[2],
+                    "had_user_interaction": bool(row[3]),
+                    "last_seen_isodate": convert_unix_to_iso(row[2]),
+                    "domain": domain,
+                    "path": path,
+                }
+                for index, column in enumerate(optional_columns, start=4):
+                    field = {
+                        "mostRecentUserInteractionTime": (
+                            "most_recent_user_interaction_time"
+                        ),
+                        "mostRecentWebPushInteractionTime": (
+                            "most_recent_web_push_interaction_time"
+                        ),
+                    }[column]
+                    timestamp = row[index]
+                    result[field] = timestamp
+                    if timestamp is not None and timestamp >= 0:
+                        result[f"{field}_isodate"] = convert_unix_to_iso(timestamp)
+                self.results.append(result)
         finally:
             cur.close()
             conn.close()

src/mvt/ios/modules/mixed/webkit_session_resource_log.py

@@ -39,7 +39,7 @@ class WebkitSessionResourceLog(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,
@@ -50,7 +50,7 @@ class WebkitSessionResourceLog(IOSExtraction):
             results=results,
         )
 
-        self.results: dict = {}
+        self.results: dict = results if results is not None else {}
 
     @staticmethod
     def _extract_domains(entries):
@@ -77,14 +77,21 @@ class WebkitSessionResourceLog(IOSExtraction):
                     entry["redirect_destination"]
                 )
 
-                # TODO: Currently not used.
-                # subframe_origins = self._extract_domains(
-                #    entry["subframe_under_origin"])
-                # subresource_domains = self._extract_domains(
-                #    entry["subresource_under_origin"])
+                subframe_origins = self._extract_domains(
+                    entry["subframe_under_origin"]
+                )
+                subresource_domains = self._extract_domains(
+                    entry["subresource_under_origin"]
+                )
 
                 all_origins = list(
-                    set([entry["origin"]] + source_domains + destination_domains)
+                    set(
+                        [entry["origin"]]
+                        + source_domains
+                        + destination_domains
+                        + subframe_origins
+                        + subresource_domains
+                    )
                 )
 
                 ioc_match = self.indicators.check_urls(all_origins)

src/mvt/ios/modules/mixed/whatsapp.py

@@ -33,7 +33,7 @@ class Whatsapp(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,

src/mvt/ios/modules/net_base.py

@@ -30,7 +30,7 @@ class NetBase(IOSExtraction):
         results_path: Optional[str] = None,
         module_options: Optional[dict] = None,
         log: logging.Logger = logging.getLogger(__name__),
-        results: ModuleResults = [],
+        results: Optional[ModuleResults] = None,
     ) -> None:
         super().__init__(
             file_path=file_path,
@@ -322,14 +322,11 @@ class NetBase(IOSExtraction):
         self.results = sorted(self.results, key=operator.itemgetter("first_isodate"))
 
     def check_indicators(self) -> None:
-        # Check for manipulated process records.
-        # TODO: Catching KeyError for live_isodate for retro-compatibility.
-        #       This is not very good.
-        try:
+        # check_manipulated/find_deleted require "live_isodate" and
+        # "live_proc_id" keys which may be absent in older result formats.
+        if self.results and "live_isodate" in self.results[0]:
             self.check_manipulated()
             self.find_deleted()
-        except KeyError:
-            pass
 
         if not self.indicators:
             return

tests/android/test_artifact_dumpsys_battery_daily.py

@@ -69,3 +69,65 @@ class TestDumpsysBatteryDailyArtifact:
         assert downgrade_alert.event["package_name"] == "com.example.app"
         assert downgrade_alert.event["action"] == "downgrade"
         assert downgrade_alert.event["previous_vers"] == "10"
+
+    def test_newest_first_update_is_not_reported_as_downgrade(self):
+        dba = DumpsysBatteryDailyArtifact()
+        dba.parse(
+            """
+  Daily from 2026-01-10 to 2026-01-11:
+      Update com.example.app vers=102
+  Daily from 2026-01-05 to 2026-01-06:
+      Update com.example.app vers=101
+"""
+        )
+
+        assert len(dba.results) == 2
+        assert len(dba.alertstore.alerts) == 0
+        assert all(result["action"] == "update" for result in dba.results)
+
+    def test_newest_first_downgrade_creates_medium_alert(self):
+        dba = DumpsysBatteryDailyArtifact()
+        dba.parse(
+            """
+  Daily from 2026-01-10 to 2026-01-11:
+      Update com.example.app vers=101
+  Daily from 2026-01-05 to 2026-01-06:
+      Update com.example.app vers=102
+"""
+        )
+
+        assert len(dba.results) == 2
+        assert len(dba.alertstore.alerts) == 1
+
+        downgrade_alert = dba.alertstore.alerts[0]
+        assert downgrade_alert.level == AlertLevel.MEDIUM
+        assert downgrade_alert.message == (
+            "Detected downgrade of package com.example.app from vers 102 to vers 101"
+        )
+        assert downgrade_alert.event_time == "2026-01-10"
+        assert downgrade_alert.event["package_name"] == "com.example.app"
+        assert downgrade_alert.event["action"] == "downgrade"
+        assert downgrade_alert.event["previous_vers"] == "102"
+
+    def test_reinstall_after_uninstall_is_not_reported_as_downgrade(self):
+        dba = DumpsysBatteryDailyArtifact()
+        dba.parse(
+            """
+  Daily from 2026-01-15 to 2026-01-16:
+      Update com.example.app vers=10
+  Daily from 2026-01-10 to 2026-01-11:
+      Update com.example.app vers=0
+  Daily from 2026-01-05 to 2026-01-06:
+      Update com.example.app vers=102
+"""
+        )
+
+        assert len(dba.results) == 3
+        assert len(dba.alertstore.alerts) == 1
+
+        uninstall_alert = dba.alertstore.alerts[0]
+        assert uninstall_alert.level == AlertLevel.MEDIUM
+        assert uninstall_alert.message == (
+            "Detected uninstall of package com.example.app (vers 0)"
+        )
+        assert uninstall_alert.event_time == "2026-01-10"

tests/android/test_backup_parser.py

@@ -5,12 +5,24 @@
 
 import hashlib
 
-from mvt.android.parsers.backup import parse_backup_file, parse_tar_for_sms
+from mvt.android.parsers.backup import (
+    parse_ab_header,
+    parse_backup_file,
+    parse_tar_for_sms,
+)
 
 from ..utils import get_artifact
 
 
 class TestBackupParsing:
+    def test_parse_incomplete_header(self):
+        assert parse_ab_header(b"ANDROID BACKUP\n") == {
+            "backup": False,
+            "compression": None,
+            "version": None,
+            "encryption": None,
+        }
+
     def test_parsing_noencryption(self):
         file = get_artifact("android_backup/backup.ab")
         with open(file, "rb") as f: