mirror of
https://github.com/mvt-project/mvt.git
synced 2026-10-22 23:28:23 +02:00
Compare commits
7
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
9d431acc77 | ||
|
|
1cf82f44e9 | ||
|
|
da33c2fdb3 | ||
|
|
1b163749b2 | ||
|
|
70efefde22 | ||
|
|
b3585e76ad | ||
|
|
1e45ead1a5 |
No files matched your search
@@ -60,6 +60,8 @@ For alternative installation options and known issues, please refer to the [docu
|
||||
|
||||
MVT provides three commands: `mvt-ios` and `mvt-android` analyse acquisitions from devices of that platform, and `mvt` hosts what belongs to neither: `version`, `completion`, `plugins` and `download-iocs` (`version` and `download-iocs` remain available on the platform commands for now). Running `mvt` on its own shows the installed version, update notices and the available commands. [Check out the documentation to learn how to use them!](https://docs.mvt.re/)
|
||||
|
||||
Pass `--verbose` to any of the three commands, before the command name (`mvt-ios --verbose check-backup ...`), for debug output. The `--verbose` option the `check-*` commands accept after their name still works but is kept for compatibility only and will be removed in a future release.
|
||||
|
||||
### Shell completion
|
||||
|
||||
MVT can generate a shell completion script for Bash, Zsh, and Fish which covers `mvt`, `mvt-ios` and `mvt-android`:
|
||||
|
||||
@@ -58,7 +58,7 @@ mvt-android check-intrusion-logs --output /path/to/results/ /path/to/intrusion-l
|
||||
| `-l, --list-modules` | List the available intrusion-log modules and exit. |
|
||||
| `-m, --module NAME` | Run a single module (e.g. `DnsEvent`) instead of all of them. |
|
||||
| `-t, --timezone TZ` | IANA timezone name for the device (e.g. `Europe/Paris`). When set, event timestamps are converted to the device's local time instead of UTC. |
|
||||
| `-v, --verbose` | Verbose logging. |
|
||||
| `-v, --verbose` | Verbose logging. Kept for compatibility and to be removed in a future release: pass `--verbose` to `mvt-android` itself instead. |
|
||||
|
||||
## Modules
|
||||
|
||||
|
||||
@@ -29,6 +29,10 @@ def summarize(path):
|
||||
click.echo(f"Summarizing {path}")
|
||||
```
|
||||
|
||||
Log through `get_plugin_logger(__name__)` from `mvt.plugin`. Records logged
|
||||
through `logging.getLogger(__name__)` do not reach `command.log`, and MVT's
|
||||
console handler does not show them.
|
||||
|
||||
Register the object in the package's `pyproject.toml`. The entry-point name is
|
||||
the command users invoke:
|
||||
|
||||
@@ -125,6 +129,64 @@ export MVT_IOS_CUSTOM_COMMANDS=./ios_commands
|
||||
export MVT_ANDROID_CUSTOM_COMMANDS=./android_commands
|
||||
```
|
||||
|
||||
## Building a Module-Running Command
|
||||
|
||||
A command which runs forensic modules over an acquisition subclasses `Command`.
|
||||
`Command` creates the output folder and writes `command.log`. It orders the
|
||||
modules, resolves their dependencies and runs them. It writes the result files,
|
||||
`alerts.json` and `info.json`. The subclass sets `platform`, `name` and
|
||||
`modules`:
|
||||
|
||||
```python
|
||||
from mvt.plugin import Command, MVTModule, convert_unix_to_iso, get_plugin_logger
|
||||
|
||||
log = get_plugin_logger(__name__)
|
||||
|
||||
|
||||
class APKManifest(MVTModule):
|
||||
supported_commands = (("android", "check-apks"),)
|
||||
|
||||
def run(self):
|
||||
self.results = [{"checked_at": convert_unix_to_iso(0)}]
|
||||
|
||||
|
||||
class CmdCheckAPKs(Command):
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
self.platform = "android"
|
||||
self.name = "check-apks"
|
||||
self.modules = [APKManifest]
|
||||
```
|
||||
|
||||
`platform` and `name` are the pair the modules declare in
|
||||
`supported_commands`. `modules` lists the module classes the command runs,
|
||||
imported directly. A command which runs modules of another plugin depends on
|
||||
that package in `pyproject.toml` and imports them the same way. `init()`,
|
||||
`module_init(module)` and `finish()` are optional hooks. `run()` calls them
|
||||
before the run, before each module and after the run.
|
||||
|
||||
Wrap the command in a Click command with an `--output` option. Run it, then
|
||||
print the alert summary:
|
||||
|
||||
```python
|
||||
import click
|
||||
|
||||
|
||||
@click.command("check-apks")
|
||||
@click.option("--output", "-o", type=click.Path(exists=False))
|
||||
@click.argument("TARGET_PATH", type=click.Path(exists=True))
|
||||
def cli(output, target_path):
|
||||
cmd = CmdCheckAPKs(target_path=target_path, results_path=output)
|
||||
log.info("Checking APK files at path: %s", target_path)
|
||||
cmd.run()
|
||||
cmd.show_alerts_brief()
|
||||
```
|
||||
|
||||
The `--verbose` option of `mvt`, `mvt-ios` and `mvt-android` applies to the
|
||||
command. The command defines no `--verbose` option of its own. The pair a
|
||||
plugin command adds is not listed anywhere in MVT. Name it in the plugin's
|
||||
README.
|
||||
|
||||
## Naming and Errors
|
||||
|
||||
Built-in MVT commands cannot be replaced. External command names must also be
|
||||
|
||||
+211
-56
@@ -48,51 +48,21 @@ configuration problem: the command logs a warning and runs no modules at all.
|
||||
|
||||
## 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:
|
||||
MVT's module-running `check-*` commands can run forensic modules which are not
|
||||
part of MVT. Custom modules are distributed as plugin packages. A Python
|
||||
package installed next to MVT registers its modules through an entry point.
|
||||
The modules then load automatically in every command they support, see
|
||||
[Installed module packages](#installed-module-packages). `mvt plugins list`
|
||||
shows the installed packages and where each one was installed from.
|
||||
|
||||
```bash
|
||||
mvt-ios check-backup --load-module ./example_module.py --output ./out ./backup
|
||||
```
|
||||
MVT can also load module files by path, with `--load-module` and
|
||||
`MVT_CUSTOM_MODULES`, see
|
||||
[Developing modules locally](#developing-modules-locally). This can be used
|
||||
while writing a module. Use a Python package to distribute one.
|
||||
|
||||
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 declare the platform/command pairs a module
|
||||
supports. Empty `supported_commands` means the module will not run and MVT logs
|
||||
a warning. This explicit declaration is required for every command. Supported
|
||||
pairs are:
|
||||
A custom module declares in `supported_commands` the platform and command
|
||||
pairs it runs in. A module with empty `supported_commands` does not run and
|
||||
MVT logs a warning. The nine pairs are:
|
||||
|
||||
```python
|
||||
("ios", "check-backup")
|
||||
@@ -106,13 +76,86 @@ pairs are:
|
||||
("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:
|
||||
`check-iocs` re-checks stored results rather than an acquisition. It matches
|
||||
every `<slug>.json` file in the results folder to the module with that slug.
|
||||
It then runs that module's `check_indicators()` again.
|
||||
|
||||
### Writing a module
|
||||
|
||||
A module subclasses `MVTModule` or one of the base classes below and
|
||||
implements `run()`. `check_indicators()` and `serialize()` are optional. The
|
||||
first matches results against IOCs or detections. The second returns timeline
|
||||
records.
|
||||
|
||||
```python
|
||||
from mvt.common.module import MVTModule
|
||||
from mvt.ios.modules.backup.manifest import Manifest
|
||||
from mvt.plugin import IOSExtraction, convert_unix_to_iso
|
||||
|
||||
|
||||
class ExampleCustomModule(IOSExtraction):
|
||||
supported_commands = (
|
||||
("ios", "check-backup"),
|
||||
("ios", "check-fs"),
|
||||
)
|
||||
slug = "example_custom_module"
|
||||
|
||||
def run(self):
|
||||
self.results = [{"checked_at": convert_unix_to_iso(0)}]
|
||||
|
||||
def check_indicators(self):
|
||||
pass
|
||||
|
||||
def serialize(self, result):
|
||||
return None
|
||||
```
|
||||
|
||||
The base classes are:
|
||||
|
||||
- `MVTModule`: the base of every module. It provides `self.results`,
|
||||
`self.alertstore`, `self.log`, `self.indicators` and
|
||||
`get_dependency_results()`. Subclass it directly for a module which reads
|
||||
only the results of other modules.
|
||||
- `IOSExtraction`: `("ios", "check-backup")` and `("ios", "check-fs")`. Adds
|
||||
`_find_ios_database()`, which locates a module's database in a backup or in
|
||||
a filesystem dump and repairs it if it is malformed. Adds
|
||||
`_get_backup_files_from_manifest()`, `_get_backup_file_from_id()` and
|
||||
`_get_fs_files_from_patterns()`. Adds `_open_sqlite_db()`, which opens a
|
||||
database read-only.
|
||||
- `SysdiagnoseExtraction`: `("ios", "check-sysdiagnose")`. MVT extracts the
|
||||
archive and calls `from_sysdiagnose_folder()` before `run()`. The module
|
||||
reads files with `_get_files_by_pattern()` and `_get_file_content()`.
|
||||
`ips_files` lists the crash reports. See
|
||||
[Check an iOS Sysdiagnose](../ios/sysdiagnose.md).
|
||||
- `AndroidQFModule`: `("android", "check-androidqf")`. MVT calls `from_dir()`
|
||||
or `from_zip()` with the file list of the acquisition. The module reads
|
||||
files with `_get_files_by_pattern()` and `_get_file_content()`.
|
||||
`_get_device_timezone()` returns the device timezone.
|
||||
- `AndroidBackupModule`: `("android", "check-backup")`. MVT calls `from_dir()`
|
||||
or `from_ab()`. The module reads files with `_get_files_by_pattern()` and
|
||||
`_get_file_content()`.
|
||||
- `BugReportModule`: `("android", "check-bugreport")`. MVT calls `from_dir()`
|
||||
or `from_zip()`. The module reads files with `_get_files_by_pattern()`,
|
||||
`_get_files_by_patterns()` and `_get_file_content()`.
|
||||
`_get_dumpstate_file()` returns the dumpstate file, and
|
||||
`_get_file_modification_time()` the modification time of a file.
|
||||
|
||||
The underscore-named helpers are internal to the base classes. Plugin modules
|
||||
can call them. Their names and signatures can change between releases. Read
|
||||
the base class in `src/mvt/ios/modules` or `src/mvt/android/modules` before
|
||||
relying on one.
|
||||
|
||||
### Depending on a built-in module
|
||||
|
||||
A module which post-processes records generated by one or more built-in MVT
|
||||
modules must declare the source modules in `dependencies`. It reads their
|
||||
results with `get_dependency_results()`. Import the class from its family
|
||||
package: `mvt.ios.modules.backup`, `mvt.ios.modules.fs`,
|
||||
`mvt.ios.modules.mixed`, `mvt.android.modules.androidqf`,
|
||||
`mvt.android.modules.backup`, `mvt.android.modules.bugreport` or
|
||||
`mvt.android.modules.intrusion_logs`.
|
||||
|
||||
```python
|
||||
from mvt.ios.modules.backup import Manifest
|
||||
from mvt.plugin import MVTModule
|
||||
|
||||
|
||||
class DependentCustomModule(MVTModule):
|
||||
@@ -124,6 +167,83 @@ class DependentCustomModule(MVTModule):
|
||||
self.results = [{"manifest_entries": len(manifest_results)}]
|
||||
```
|
||||
|
||||
Dependencies are ordered as for the built-in modules, with custom modules
|
||||
appended after the built-ins, see [Module dependencies](#module-dependencies).
|
||||
A dependency has to run in every command the module supports. Where it does
|
||||
not, MVT skips the module with a warning.
|
||||
|
||||
`get_dependency_results()` returns the plain dictionaries the module produced.
|
||||
They are the same records it writes to `<slug>.json`. Typed results per module
|
||||
are planned.
|
||||
|
||||
### Replacing a built-in module
|
||||
|
||||
A custom module which extends a built-in one runs alongside it, and when the
|
||||
two share a slug they write to the same results file. MVT warns about that
|
||||
collision, naming both modules, where each came from and the file the later one
|
||||
overwrites, but it runs them both. Set `replaces` to the class the module
|
||||
supersedes to take that module's place instead: when both are available to a
|
||||
command, the named module is dropped from the run.
|
||||
|
||||
```python
|
||||
from mvt.ios.modules.backup import Manifest as BuiltinManifest
|
||||
|
||||
|
||||
class Manifest(BuiltinManifest):
|
||||
supported_commands = (("ios", "check-backup"),)
|
||||
replaces = BuiltinManifest
|
||||
```
|
||||
|
||||
Keeping the class name of the replaced module, as above, keeps the name
|
||||
`--module` selects the module by and the slug its results file is named after.
|
||||
A replacement with a different class name writes to a file named after its own
|
||||
slug, unless it sets `slug` to the slug of the module it replaces, and
|
||||
`--module` still selects it by the name of the replaced module, which MVT logs.
|
||||
Taking over the slug of a replaced module is not a collision and is not warned
|
||||
about, because that module is no longer part of the run.
|
||||
|
||||
Modules which depend on the replaced module receive the replacement instead, so
|
||||
`get_dependency_results(BuiltinManifest)` returns the replacement's results. A
|
||||
module must not depend on the module it replaces: that module does not run, so
|
||||
the replacement has to produce the data itself, and MVT logs a warning when a
|
||||
module declares both.
|
||||
|
||||
A replacement takes on the obligations of the module it replaces. If it
|
||||
declares a dependency the command does not provide, it is skipped like any
|
||||
other module with an unavailable dependency, and the module it replaced stays
|
||||
out of the run: neither of them produces results, and the modules depending on
|
||||
the replaced module are skipped as well.
|
||||
|
||||
Subclassing the replaced module is not required, but a replacement which is not
|
||||
a subclass is logged with a warning, because its results may not be what the
|
||||
modules depending on the replaced module expect. Naming a module the command
|
||||
does not run has no effect, and modules which replace each other in a cycle all
|
||||
keep running and replace nothing. Every applied substitution is logged, so it
|
||||
is recorded in `command.log` when the command runs with an `--output` folder.
|
||||
|
||||
Every command resolves replacements on its own. `check-iocs` matches stored
|
||||
results files against the slugs of the modules available for that command, so a
|
||||
replacement checks the indicators of its own results only if it also declares
|
||||
the `("ios", "check-iocs")` pair; otherwise the built-in module it replaced
|
||||
re-checks the file. It also matches `--module` on the class name only, so pass
|
||||
a differently named replacement's own name there, not the name of the module
|
||||
it replaces.
|
||||
|
||||
### Importing from MVT
|
||||
|
||||
Import from `mvt.plugin` if it has what you need. The names it exports are
|
||||
kept working on a best-effort basis. A change to one of them is announced in
|
||||
the release notes. Anything else in `mvt` can be imported too, but may change
|
||||
between releases without notice. The plugin interface is best effort.
|
||||
|
||||
`mvt.plugin` exports the base classes above and `Command`, `Alert` and
|
||||
`AlertLevel`, the result types, `DatabaseNotFoundError` and
|
||||
`DatabaseCorruptedError`, the timestamp converters, the settings API of
|
||||
[Plugin Configuration](plugin_configuration.md), MVT's own `settings`,
|
||||
`get_plugin_logger()` and `MVT_VERSION`. `src/mvt/plugin.py` holds the list.
|
||||
Read MVT's `settings` for values such as `NETWORK_ACCESS_ALLOWED` and
|
||||
`NETWORK_TIMEOUT`. Plugin values go in the plugin's own settings file.
|
||||
|
||||
## Installed module packages
|
||||
|
||||
Python packages can register modules so they load automatically in every
|
||||
@@ -133,14 +253,14 @@ the package's `pyproject.toml`:
|
||||
|
||||
```toml
|
||||
[project.entry-points."mvt.modules"]
|
||||
mvt-plugin-amnesty-custom = "mvt_plugin_amnesty_custom:get_modules"
|
||||
mvt-plugin-example-org = "mvt_plugin_example_org:get_modules"
|
||||
```
|
||||
|
||||
The entry point must resolve to an iterable of `MVTModule` subclasses, or to
|
||||
a callable returning one:
|
||||
|
||||
```python
|
||||
from mvt.common.module import MVTModule
|
||||
from mvt.plugin import MVTModule
|
||||
|
||||
|
||||
class PackagedModule(MVTModule):
|
||||
@@ -154,6 +274,10 @@ def get_modules() -> list[type[MVTModule]]:
|
||||
return [PackagedModule]
|
||||
```
|
||||
|
||||
`get_modules()` is the package's module list, written by hand. A package which
|
||||
keeps its modules in separate files imports each class there and lists it. A
|
||||
module missing from the list does not load.
|
||||
|
||||
Installed modules follow the same rules as other custom modules: each module
|
||||
must declare `supported_commands`, and dependencies are resolved with the
|
||||
standard ordering logic. A broken entry point is skipped with a warning and
|
||||
@@ -164,7 +288,7 @@ from sources you trust.
|
||||
For a `pipx` installation of MVT, inject the package into MVT's environment:
|
||||
|
||||
```bash
|
||||
pipx inject mvt mvt-plugin-amnesty-custom
|
||||
pipx inject mvt mvt-plugin-example-org
|
||||
```
|
||||
|
||||
Module packages that need their own settings, such as an API key, should store
|
||||
@@ -175,9 +299,9 @@ rather than in MVT's own `config.yaml`.
|
||||
|
||||
Name module packages `mvt-plugin-<name>` (import package `mvt_plugin_<name>`),
|
||||
and include the name of the publishing organization or author so packages from
|
||||
different groups do not collide: for example, Amnesty International's custom
|
||||
modules would be distributed as `mvt-plugin-amnesty-custom` with the import
|
||||
package `mvt_plugin_amnesty_custom`.
|
||||
different groups do not collide: for example, an organisation's custom modules
|
||||
would be distributed as `mvt-plugin-example-org` with the import package
|
||||
`mvt_plugin_example_org`.
|
||||
|
||||
The prefix makes module packages easy to find on PyPI and keeps their import
|
||||
names from clashing with unrelated Python packages. It is a convention, not a
|
||||
@@ -196,11 +320,42 @@ came from. MVT's own modules log under their dotted path (for example
|
||||
MVT's internal logger tree:
|
||||
|
||||
- Installed packages log under `mvt.ext.<package>`, with the `mvt_plugin_`
|
||||
prefix stripped: modules in `mvt_plugin_amnesty_custom` log as
|
||||
`mvt.ext.amnesty_custom.*`.
|
||||
prefix stripped: modules in `mvt_plugin_example_org` log as
|
||||
`mvt.ext.example_org.*`.
|
||||
- Files loaded with `--load-module` or `MVT_CUSTOM_MODULES` log as
|
||||
`mvt.ext.<file name>`.
|
||||
|
||||
Outside a module class, for example in a custom command line handler, log
|
||||
through `get_plugin_logger(__name__)`. It returns a logger in the same
|
||||
namespace.
|
||||
|
||||
## Developing modules locally
|
||||
|
||||
While a module is being written, load it from its file. `--load-module` takes a
|
||||
Python file, or a folder of them, on every module-running command, and can be
|
||||
repeated:
|
||||
|
||||
```bash
|
||||
mvt-ios check-backup --load-module ./example_module.py --output ./out ./backup
|
||||
```
|
||||
|
||||
For a folder, MVT loads its non-hidden top-level `*.py` files in sorted order
|
||||
and skips `__init__.py`. `MVT_CUSTOM_MODULES` names a folder to load on every
|
||||
module-running command, before any `--load-module` path:
|
||||
|
||||
```bash
|
||||
MVT_CUSTOM_MODULES=./custom_modules mvt-android check-bugreport ./bugreport.zip
|
||||
```
|
||||
|
||||
Files loaded this way follow the same rules as packaged modules.
|
||||
`--list-modules` reports them with the SHA-256 hash of the file in place of a
|
||||
version. An editable install of the package (`pip install -e .`) also works:
|
||||
the modules load through the entry point, and `mvt plugins list` shows the
|
||||
package with the `local` origin.
|
||||
|
||||
Loading by path is for development. Move a module into a package once it
|
||||
works.
|
||||
|
||||
## Auditing loaded modules
|
||||
|
||||
Because installed module packages load automatically, MVT records where every
|
||||
|
||||
@@ -24,8 +24,7 @@ configuration:
|
||||
|
||||
The exact parent folder follows the platform convention used for MVT's
|
||||
`config.yaml` (for example `~/Library/Application Support/mvt` on macOS). Use
|
||||
`mvt.common.plugin_config.plugin_config_path()` instead of building the path by
|
||||
hand.
|
||||
`plugin_config_path()` from `mvt.plugin` instead of building the path by hand.
|
||||
|
||||
Plugin names must be lowercase and may only contain letters, digits and dashes,
|
||||
matching the `mvt-plugin-<name>` package naming convention. MVT creates the
|
||||
@@ -38,9 +37,8 @@ leaves a partially written settings file behind.
|
||||
|
||||
Everything else a plugin keeps on disk, such as a cache, a downloaded artifact
|
||||
or synchronization state, belongs in the folder returned by the `data_folder()`
|
||||
class method of the plugin's settings class, or by
|
||||
`mvt.common.plugin_config.plugin_data_folder()` called with the plugin name if
|
||||
the plugin has no settings class:
|
||||
class method of the plugin's settings class, or by `plugin_data_folder()` from
|
||||
`mvt.plugin`, called with the plugin name if the plugin has no settings class:
|
||||
|
||||
```
|
||||
~/.local/share/mvt/plugin-data/<plugin name>/ # Linux
|
||||
@@ -76,7 +74,7 @@ defaults:
|
||||
```python
|
||||
from typing import Optional
|
||||
|
||||
from mvt.common.plugin_config import MVTPluginSettings
|
||||
from mvt.plugin import MVTPluginSettings
|
||||
|
||||
|
||||
class ExamplePluginSettings(MVTPluginSettings):
|
||||
@@ -94,12 +92,15 @@ from datetime import datetime, timezone
|
||||
|
||||
import click
|
||||
|
||||
from mvt.plugin import plugin_env_prefix
|
||||
|
||||
|
||||
def sync():
|
||||
settings = ExamplePluginSettings.load()
|
||||
if not settings.API_KEY:
|
||||
prefix = plugin_env_prefix(settings.plugin_name)
|
||||
raise click.ClickException(
|
||||
"No API key configured. Set MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY or "
|
||||
f"No API key configured. Set {prefix}API_KEY or "
|
||||
"run 'example-plugin configure'."
|
||||
)
|
||||
|
||||
@@ -107,6 +108,9 @@ def sync():
|
||||
settings.save()
|
||||
```
|
||||
|
||||
The message builds the variable name with `plugin_env_prefix()`, see
|
||||
[Environment Variables](#environment-variables).
|
||||
|
||||
A missing settings file is not an error: the plugin then runs on the field
|
||||
defaults and on whatever the environment provides. `save()` only persists the
|
||||
values that differ from the defaults, and it never touches MVT's `config.yaml`.
|
||||
@@ -129,6 +133,10 @@ export MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY=...
|
||||
export MVT_PLUGIN_EXAMPLE_PLUGIN_MAX_RESULTS=50
|
||||
```
|
||||
|
||||
Do not repeat the plugin name in a field name: a plugin named `example-scanner`
|
||||
with a `SCANNER_API_KEY` field asks the user for
|
||||
`MVT_PLUGIN_EXAMPLE_SCANNER_SCANNER_API_KEY`. Name the field `API_KEY`.
|
||||
|
||||
Settings resolve in this order, from highest to lowest priority:
|
||||
|
||||
1. Arguments passed to the settings class directly, such as
|
||||
|
||||
+18
-14
@@ -1,30 +1,32 @@
|
||||
# Check an iOS Sysdiagnose
|
||||
|
||||
`mvt-ios check-sysdiagnose` prepares an iOS sysdiagnose archive for analysis by
|
||||
custom MVT modules. MVT does not include built-in sysdiagnose modules. You must
|
||||
load at least one custom module that explicitly supports this command.
|
||||
custom MVT modules. MVT does not include built-in sysdiagnose modules. The
|
||||
command runs the modules of the installed
|
||||
[plugin packages](../development/index.md#installed-module-packages) which
|
||||
declare support for it. Install at least one such package first.
|
||||
|
||||
The command accepts either an extracted sysdiagnose directory or the original
|
||||
gzip-compressed tar archive.
|
||||
|
||||
```bash
|
||||
mvt-ios check-sysdiagnose \
|
||||
--load-module ./sysdiagnose_modules.py \
|
||||
--output ./results \
|
||||
mvt-ios check-sysdiagnose --output ./results \
|
||||
./sysdiagnose_2024.01.02_03-04-05+0200.tar.gz
|
||||
```
|
||||
|
||||
Use `--hashes` to include hashes for analyzed files in `info.json`, and
|
||||
`--list-modules` to display the eligible custom modules without running them.
|
||||
`--list-modules` to display the eligible modules without running them.
|
||||
|
||||
## Writing a custom module
|
||||
|
||||
Extend `SysdiagnoseExtraction` to access the archive contents consistently for
|
||||
both directory and tar inputs. Each module must declare the command explicitly
|
||||
in `supported_commands`.
|
||||
Extend `SysdiagnoseExtraction` from `mvt.plugin`, see
|
||||
[Writing a module](../development/index.md#writing-a-module). The module reads
|
||||
the archive the same way whether MVT was given a folder or a tar archive. It
|
||||
declares the command in `supported_commands`. While writing one,
|
||||
[load it from its file](../development/index.md#developing-modules-locally).
|
||||
|
||||
```python
|
||||
from mvt.ios.modules.sysdiagnose import SysdiagnoseExtraction
|
||||
from mvt.plugin import SysdiagnoseExtraction
|
||||
|
||||
|
||||
class ExampleSysdiagnoseModule(SysdiagnoseExtraction):
|
||||
@@ -44,7 +46,9 @@ class ExampleSysdiagnoseModule(SysdiagnoseExtraction):
|
||||
return None
|
||||
```
|
||||
|
||||
The base class provides `from_sysdiagnose_folder()` and
|
||||
`from_sysdiagnose_tar()` setup hooks, as well as protected file lookup, file
|
||||
reading, and timezone extraction helpers. IPS crash-report metadata is exposed
|
||||
on `ips_files`.
|
||||
MVT extracts a tar archive first. It calls `from_sysdiagnose_folder()` on each
|
||||
module before `run()`. `ips_files` lists the IPS crash reports.
|
||||
|
||||
`_get_files_by_pattern()` and `_get_file_content()` are internal helpers of the
|
||||
base class. Use them to read the archive. Their names and signatures can change
|
||||
between releases. See `src/mvt/ios/modules/sysdiagnose/base.py`.
|
||||
+20
-16
@@ -36,6 +36,7 @@ from mvt.common.help import (
|
||||
HELP_MSG_OUTPUT,
|
||||
HELP_MSG_STIX2,
|
||||
HELP_MSG_VERBOSE,
|
||||
HELP_MSG_VERBOSE_COMMAND,
|
||||
HELP_MSG_VERSION,
|
||||
HELP_MSG_VIRUS_TOTAL,
|
||||
)
|
||||
@@ -48,11 +49,8 @@ from .cmd_check_androidqf import CmdAndroidCheckAndroidQF
|
||||
from .cmd_check_backup import CmdAndroidCheckBackup
|
||||
from .cmd_check_bugreport import CmdAndroidCheckBugreport
|
||||
from .cmd_check_intrusion_logs import CmdAndroidCheckIntrusionLogs
|
||||
from .modules.intrusion_logs import INTRUSION_LOGS_MODULES
|
||||
from .modules.androidqf import ANDROIDQF_MODULES
|
||||
from .modules.backup import BACKUP_MODULES
|
||||
from .command_modules import ANDROID_CHECK_IOCS_MODULES
|
||||
from .modules.backup.helpers import cli_load_android_backup_password
|
||||
from .modules.bugreport import BUGREPORT_MODULES
|
||||
|
||||
init_logging()
|
||||
log = logging.getLogger("mvt")
|
||||
@@ -70,6 +68,11 @@ def _get_disable_flags(ctx):
|
||||
)
|
||||
|
||||
|
||||
def _get_verbose(ctx):
|
||||
"""Return whether --verbose was passed to the CLI itself."""
|
||||
return bool(ctx.obj and ctx.obj.get("verbose", False))
|
||||
|
||||
|
||||
def _load_custom_modules(load_module):
|
||||
try:
|
||||
return load_custom_modules(load_module)
|
||||
@@ -90,11 +93,14 @@ def _load_custom_modules(load_module):
|
||||
is_flag=True,
|
||||
help=HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK,
|
||||
)
|
||||
@click.option("--verbose", "-v", is_flag=True, help=HELP_MSG_VERBOSE)
|
||||
@click.pass_context
|
||||
def cli(ctx, disable_update_check, disable_indicator_update_check):
|
||||
def cli(ctx, disable_update_check, disable_indicator_update_check, verbose):
|
||||
ctx.ensure_object(dict)
|
||||
ctx.obj["disable_version_check"] = disable_update_check
|
||||
ctx.obj["disable_indicator_check"] = disable_indicator_update_check
|
||||
ctx.obj["verbose"] = verbose
|
||||
set_verbose_logging(verbose)
|
||||
logo(
|
||||
disable_version_check=disable_update_check,
|
||||
disable_indicator_check=disable_indicator_update_check,
|
||||
@@ -145,7 +151,7 @@ def check_adb(ctx):
|
||||
default=[],
|
||||
help=HELP_MSG_LOAD_MODULE,
|
||||
)
|
||||
@click.option("--verbose", "-v", is_flag=True, help=HELP_MSG_VERBOSE)
|
||||
@click.option("--verbose", "-v", is_flag=True, help=HELP_MSG_VERBOSE_COMMAND)
|
||||
@click.argument("BUGREPORT_PATH", type=click.Path(exists=True))
|
||||
@click.pass_context
|
||||
def check_bugreport(
|
||||
@@ -158,7 +164,7 @@ def check_bugreport(
|
||||
verbose,
|
||||
bugreport_path,
|
||||
):
|
||||
set_verbose_logging(verbose)
|
||||
set_verbose_logging(verbose or _get_verbose(ctx))
|
||||
custom_modules = _load_custom_modules(load_module)
|
||||
# Always generate hashes as bug reports are small.
|
||||
cmd = CmdAndroidCheckBugreport(
|
||||
@@ -213,7 +219,7 @@ def check_bugreport(
|
||||
)
|
||||
@click.option("--non-interactive", "-n", is_flag=True, help=HELP_MSG_NONINTERACTIVE)
|
||||
@click.option("--backup-password", "-p", help=HELP_MSG_ANDROID_BACKUP_PASSWORD)
|
||||
@click.option("--verbose", "-v", is_flag=True, help=HELP_MSG_VERBOSE)
|
||||
@click.option("--verbose", "-v", is_flag=True, help=HELP_MSG_VERBOSE_COMMAND)
|
||||
@click.argument("BACKUP_PATH", type=click.Path(exists=True))
|
||||
@click.pass_context
|
||||
def check_backup(
|
||||
@@ -227,7 +233,7 @@ def check_backup(
|
||||
verbose,
|
||||
backup_path,
|
||||
):
|
||||
set_verbose_logging(verbose)
|
||||
set_verbose_logging(verbose or _get_verbose(ctx))
|
||||
custom_modules = _load_custom_modules(load_module)
|
||||
|
||||
# Always generate hashes as backups are generally small.
|
||||
@@ -287,7 +293,7 @@ def check_backup(
|
||||
)
|
||||
@click.option("--non-interactive", "-n", is_flag=True, help=HELP_MSG_NONINTERACTIVE)
|
||||
@click.option("--backup-password", "-p", help=HELP_MSG_ANDROID_BACKUP_PASSWORD)
|
||||
@click.option("--verbose", "-v", is_flag=True, help=HELP_MSG_VERBOSE)
|
||||
@click.option("--verbose", "-v", is_flag=True, help=HELP_MSG_VERBOSE_COMMAND)
|
||||
@click.argument("ANDROIDQF_PATH", type=click.Path(exists=True))
|
||||
@click.pass_context
|
||||
def check_androidqf(
|
||||
@@ -305,7 +311,7 @@ def check_androidqf(
|
||||
verbose,
|
||||
androidqf_path,
|
||||
):
|
||||
set_verbose_logging(verbose)
|
||||
set_verbose_logging(verbose or _get_verbose(ctx))
|
||||
custom_modules = _load_custom_modules(load_module)
|
||||
|
||||
cmd = CmdAndroidCheckAndroidQF(
|
||||
@@ -373,7 +379,7 @@ def check_androidqf(
|
||||
"time instead of UTC."
|
||||
),
|
||||
)
|
||||
@click.option("--verbose", "-v", is_flag=True, help=HELP_MSG_VERBOSE)
|
||||
@click.option("--verbose", "-v", is_flag=True, help=HELP_MSG_VERBOSE_COMMAND)
|
||||
@click.argument("LOGS_PATH", type=click.Path(exists=True))
|
||||
@click.pass_context
|
||||
def check_intrusion_logs(
|
||||
@@ -387,7 +393,7 @@ def check_intrusion_logs(
|
||||
verbose,
|
||||
logs_path,
|
||||
):
|
||||
set_verbose_logging(verbose)
|
||||
set_verbose_logging(verbose or _get_verbose(ctx))
|
||||
custom_modules = _load_custom_modules(load_module)
|
||||
|
||||
module_options = {}
|
||||
@@ -450,9 +456,7 @@ def check_iocs(ctx, iocs, list_modules, module, load_module, folder):
|
||||
custom_modules=custom_modules,
|
||||
platform="android",
|
||||
)
|
||||
cmd.modules = (
|
||||
BACKUP_MODULES + BUGREPORT_MODULES + ANDROIDQF_MODULES + INTRUSION_LOGS_MODULES
|
||||
)
|
||||
cmd.modules = ANDROID_CHECK_IOCS_MODULES
|
||||
|
||||
if list_modules:
|
||||
cmd.list_modules()
|
||||
|
||||
@@ -0,0 +1,23 @@
|
||||
# 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/
|
||||
|
||||
"""Module lists an mvt-android command composes from more than one family.
|
||||
|
||||
Commands whose modules are one family read that family directly. check-iocs
|
||||
re-checks stored results, so it has to know every module that could have
|
||||
written one, and both the CLI and any other code needing that answer share
|
||||
the list from here rather than each concatenating their own.
|
||||
"""
|
||||
|
||||
from mvt.common.module import MVTModule
|
||||
|
||||
from .modules.androidqf import ANDROIDQF_MODULES
|
||||
from .modules.backup import BACKUP_MODULES
|
||||
from .modules.bugreport import BUGREPORT_MODULES
|
||||
from .modules.intrusion_logs import INTRUSION_LOGS_MODULES
|
||||
|
||||
ANDROID_CHECK_IOCS_MODULES: list[type[MVTModule]] = (
|
||||
BACKUP_MODULES + BUGREPORT_MODULES + ANDROIDQF_MODULES + INTRUSION_LOGS_MODULES
|
||||
)
|
||||
+6
-2
@@ -17,11 +17,12 @@ from mvt.common.help import (
|
||||
HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK,
|
||||
HELP_MSG_DISABLE_UPDATE_CHECK,
|
||||
HELP_MSG_STIX2,
|
||||
HELP_MSG_VERBOSE,
|
||||
HELP_MSG_VERSION,
|
||||
)
|
||||
from mvt.common.logo import logo
|
||||
from mvt.common.updates import IndicatorsUpdates
|
||||
from mvt.common.utils import init_logging
|
||||
from mvt.common.utils import init_logging, set_verbose_logging
|
||||
|
||||
init_logging()
|
||||
|
||||
@@ -41,8 +42,9 @@ CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])
|
||||
is_flag=True,
|
||||
help=HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK,
|
||||
)
|
||||
@click.option("--verbose", "-v", is_flag=True, help=HELP_MSG_VERBOSE)
|
||||
@click.pass_context
|
||||
def cli(ctx, disable_update_check, disable_indicator_update_check):
|
||||
def cli(ctx, disable_update_check, disable_indicator_update_check, verbose):
|
||||
"""Mobile Verification Toolkit.
|
||||
|
||||
mvt-ios and mvt-android run the forensic analysis of an acquisition: each
|
||||
@@ -53,6 +55,8 @@ 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
|
||||
ctx.obj["verbose"] = verbose
|
||||
set_verbose_logging(verbose)
|
||||
if ctx.invoked_subcommand != "completion":
|
||||
logo(
|
||||
disable_version_check=disable_update_check,
|
||||
|
||||
@@ -15,6 +15,8 @@ from typing import Iterable
|
||||
|
||||
import click
|
||||
|
||||
from .module_loader import CUSTOM_COMMAND_MODULE_PREFIX
|
||||
|
||||
IOS_CLI_PLUGIN_GROUP = "mvt.ios.cli_plugins"
|
||||
ANDROID_CLI_PLUGIN_GROUP = "mvt.android.cli_plugins"
|
||||
# Commands in this group are registered on the platform-neutral mvt command only.
|
||||
@@ -57,7 +59,7 @@ class BrokenPluginCommand(click.Command):
|
||||
|
||||
def _module_name_for_path(path: Path) -> str:
|
||||
digest = hashlib.sha256(str(path).encode("utf-8")).hexdigest()[:16]
|
||||
return f"_mvt_custom_command_{path.stem}_{digest}"
|
||||
return f"{CUSTOM_COMMAND_MODULE_PREFIX}{path.stem}_{digest}"
|
||||
|
||||
|
||||
def _iter_command_files(path: Path) -> Iterable[Path]:
|
||||
|
||||
+265
-16
@@ -72,6 +72,10 @@ class Command:
|
||||
# down a password to decrypt a backup or flags which are need by some modules.
|
||||
self.module_options = module_options if module_options else {}
|
||||
|
||||
# This dictionary maps the modules which were replaced by a module
|
||||
# declaring `replaces` to the module which took their place.
|
||||
self.module_replacements: dict[type[MVTModule], type[MVTModule]] = {}
|
||||
|
||||
# This list will contain all executed modules.
|
||||
# We can use this to reference e.g. self.executed[0].results.
|
||||
self.executed: list[MVTModule] = []
|
||||
@@ -257,7 +261,174 @@ class Command:
|
||||
if module not in deduplicated:
|
||||
deduplicated.append(module)
|
||||
|
||||
return deduplicated
|
||||
available = self._apply_replacements(deduplicated)
|
||||
self._warn_about_slug_collisions(available)
|
||||
|
||||
return available
|
||||
|
||||
def _warn_about_slug_collisions(self, modules: list[type[MVTModule]]) -> None:
|
||||
"""Report modules writing their results to the same file.
|
||||
|
||||
Results are stored in a file named after the module slug, so two
|
||||
modules sharing one silently overwrite each other. Replacements are
|
||||
already resolved here, so a module deliberately taking over the slug
|
||||
of the module it replaces is not reported: the module it replaced is
|
||||
no longer part of the run.
|
||||
"""
|
||||
modules_by_slug: dict[str, type[MVTModule]] = {}
|
||||
for module in modules:
|
||||
slug = module.get_slug()
|
||||
first = modules_by_slug.setdefault(slug, module)
|
||||
if first is module:
|
||||
continue
|
||||
|
||||
self.log.warning(
|
||||
"Modules %s from %s and %s from %s both use the slug %s. If "
|
||||
"both run, whichever runs last overwrites the results of the "
|
||||
"other in %s.json.",
|
||||
first.__name__,
|
||||
get_module_origin(first).label,
|
||||
module.__name__,
|
||||
get_module_origin(module).label,
|
||||
slug,
|
||||
slug,
|
||||
)
|
||||
|
||||
def _declared_replacements(
|
||||
self, modules: list[type[MVTModule]]
|
||||
) -> dict[type[MVTModule], type[MVTModule]]:
|
||||
"""Return the replacements declared by the given modules."""
|
||||
replacements: dict[type[MVTModule], type[MVTModule]] = {}
|
||||
for module in modules:
|
||||
replaced = module.replaces
|
||||
if replaced is None or replaced is module:
|
||||
continue
|
||||
|
||||
if not module.enabled:
|
||||
# Replacing a module must not disable it, as a disabled
|
||||
# replacement never runs in its place.
|
||||
self.log.debug(
|
||||
"Module %s is disabled and does not replace module %s.",
|
||||
module.__name__,
|
||||
replaced.__name__,
|
||||
)
|
||||
continue
|
||||
|
||||
if replaced not in modules:
|
||||
# A module can support several commands while the module it
|
||||
# replaces is only available in some of them.
|
||||
self.log.debug(
|
||||
"Module %s replaces module %s, which is not available "
|
||||
"for the %s command.",
|
||||
module.__name__,
|
||||
replaced.__name__,
|
||||
self.name,
|
||||
)
|
||||
continue
|
||||
|
||||
if replaced in replacements:
|
||||
self.log.warning(
|
||||
"Modules %s and %s both replace module %s. Both of them "
|
||||
"will run, %s will not, and modules depending on %s will "
|
||||
"use the results of %s. Replacements which share the slug "
|
||||
"of %s overwrite each other's results file.",
|
||||
replacements[replaced].__name__,
|
||||
module.__name__,
|
||||
replaced.__name__,
|
||||
replaced.__name__,
|
||||
replaced.__name__,
|
||||
replacements[replaced].__name__,
|
||||
replaced.__name__,
|
||||
)
|
||||
continue
|
||||
|
||||
replacements[replaced] = module
|
||||
|
||||
return replacements
|
||||
|
||||
def _drop_replacement_cycles(
|
||||
self, replacements: dict[type[MVTModule], type[MVTModule]]
|
||||
) -> None:
|
||||
"""Undo the replacements between modules which replace each other."""
|
||||
cyclic: set[type[MVTModule]] = set()
|
||||
for replaced in replacements:
|
||||
walked: list[type[MVTModule]] = []
|
||||
module = replaced
|
||||
while module in replacements and module not in cyclic:
|
||||
if module in walked:
|
||||
cyclic.update(walked[walked.index(module) :])
|
||||
break
|
||||
walked.append(module)
|
||||
module = replacements[module]
|
||||
|
||||
if not cyclic:
|
||||
return
|
||||
|
||||
self.log.warning(
|
||||
"Modules %s replace each other in a cycle. None of them replaces "
|
||||
"anything and all of them will run.",
|
||||
", ".join(sorted(module.__name__ for module in cyclic)),
|
||||
)
|
||||
for module in cyclic:
|
||||
replacements.pop(module, None)
|
||||
|
||||
def _log_replacement(
|
||||
self,
|
||||
replaced: type[MVTModule],
|
||||
module: type[MVTModule],
|
||||
replacement: type[MVTModule],
|
||||
) -> None:
|
||||
"""Report an applied replacement, and any problem with it."""
|
||||
if not issubclass(module, replaced):
|
||||
self.log.warning(
|
||||
"Module %s replaces module %s but is not a subclass of it. "
|
||||
"Its results might not be compatible with what modules "
|
||||
"depending on %s expect.",
|
||||
module.__name__,
|
||||
replaced.__name__,
|
||||
replaced.__name__,
|
||||
)
|
||||
|
||||
if replaced in module.dependencies:
|
||||
self.log.warning(
|
||||
"Module %s depends on module %s, which it also replaces. The "
|
||||
"dependency cannot be satisfied: a replacement has to produce "
|
||||
"that data itself.",
|
||||
module.__name__,
|
||||
replaced.__name__,
|
||||
)
|
||||
|
||||
self.log.info(
|
||||
"Module %s from %s replaces module %s from %s.",
|
||||
replacement.__name__,
|
||||
get_module_origin(replacement).label,
|
||||
replaced.__name__,
|
||||
get_module_origin(replaced).label,
|
||||
)
|
||||
|
||||
def _apply_replacements(
|
||||
self, modules: list[type[MVTModule]]
|
||||
) -> list[type[MVTModule]]:
|
||||
"""Drop the modules superseded by a module declaring `replaces`."""
|
||||
declared = self._declared_replacements(modules)
|
||||
self._drop_replacement_cycles(declared)
|
||||
|
||||
replacements: dict[type[MVTModule], type[MVTModule]] = {}
|
||||
for replaced, module in declared.items():
|
||||
# Follow chains of replacements, so that a dependency on a
|
||||
# replaced module always resolves to a module which is part of
|
||||
# the run.
|
||||
replacement = module
|
||||
while replacement in declared:
|
||||
replacement = declared[replacement]
|
||||
|
||||
# Only the replacements which are applied are reported, so that
|
||||
# the record matches the modules which actually run.
|
||||
self._log_replacement(replaced, module, replacement)
|
||||
replacements[replaced] = replacement
|
||||
|
||||
self.module_replacements = replacements
|
||||
return [module for module in modules if module not in replacements]
|
||||
|
||||
def init(self) -> None:
|
||||
raise NotImplementedError
|
||||
@@ -317,6 +488,80 @@ class Command:
|
||||
console.print("")
|
||||
console.print(panel)
|
||||
|
||||
def _module_dependencies(
|
||||
self, module: type[MVTModule]
|
||||
) -> list[tuple[type[MVTModule], type[MVTModule]]]:
|
||||
"""Return the (declared, resolved) dependencies of a module.
|
||||
|
||||
A dependency on a module which was replaced is resolved to the module
|
||||
which took its place. A module which replaces one of its own
|
||||
dependencies is not made to depend on itself, while a module which
|
||||
declares itself as a dependency is left alone and still fails the
|
||||
circular dependency check.
|
||||
"""
|
||||
dependencies = []
|
||||
for dependency in module.dependencies:
|
||||
resolved = self.module_replacements.get(dependency, dependency)
|
||||
if resolved is module and dependency is not module:
|
||||
continue
|
||||
dependencies.append((dependency, resolved))
|
||||
|
||||
return dependencies
|
||||
|
||||
@staticmethod
|
||||
def _dependency_name(
|
||||
declared: type[MVTModule], resolved: type[MVTModule]
|
||||
) -> str:
|
||||
"""Return how a dependency is named in the messages about it.
|
||||
|
||||
A dependency is named as the module which declared it wrote it, and,
|
||||
when that module was replaced, as the module which runs in its place.
|
||||
"""
|
||||
if declared is resolved:
|
||||
return declared.__name__
|
||||
|
||||
return f"{resolved.__name__} (replacing module {declared.__name__})"
|
||||
|
||||
def _selected_modules(
|
||||
self, modules: list[type[MVTModule]]
|
||||
) -> Optional[list[type[MVTModule]]]:
|
||||
"""Return the modules explicitly requested, or all the enabled ones.
|
||||
|
||||
Returns None when a module was requested by name and no module of
|
||||
that name can be run.
|
||||
"""
|
||||
if not self.module_name:
|
||||
return [module for module in modules if module.enabled]
|
||||
|
||||
selected = [
|
||||
module for module in modules if module.__name__ == self.module_name
|
||||
]
|
||||
|
||||
# A module replacing another one does not have to keep its name, so
|
||||
# the name of a replaced module selects its replacement.
|
||||
if not selected:
|
||||
for replaced, replacement in self.module_replacements.items():
|
||||
if replaced.__name__ != self.module_name or replacement in selected:
|
||||
continue
|
||||
self.log.info(
|
||||
"Module %s was replaced by module %s, which is run "
|
||||
"in its place.",
|
||||
replaced.__name__,
|
||||
replacement.__name__,
|
||||
)
|
||||
selected.append(replacement)
|
||||
|
||||
if not selected:
|
||||
self.log.warning(
|
||||
"No module named %s is available for the %s command. "
|
||||
"No modules will be run.",
|
||||
self.module_name,
|
||||
self.name,
|
||||
)
|
||||
return None
|
||||
|
||||
return selected
|
||||
|
||||
def _skipped_modules(
|
||||
self,
|
||||
required: list[type[MVTModule]],
|
||||
@@ -349,15 +594,18 @@ class Command:
|
||||
if module in skipped:
|
||||
continue
|
||||
|
||||
for dependency in module.dependencies:
|
||||
for declared, dependency in self._module_dependencies(module):
|
||||
if dependency not in module_indexes:
|
||||
skipped[module] = (module, dependency)
|
||||
# A replaced dependency always resolves to a module of
|
||||
# this command, so an unavailable one is always the
|
||||
# module class the author declared.
|
||||
skipped[module] = (module, declared)
|
||||
changed = True
|
||||
self.log.warning(
|
||||
"Module %s will be SKIPPED: it depends on module "
|
||||
"%s, which is not available in this command.%s",
|
||||
module.__name__,
|
||||
dependency.__name__,
|
||||
declared.__name__,
|
||||
remainder,
|
||||
)
|
||||
break
|
||||
@@ -372,7 +620,7 @@ class Command:
|
||||
"module %s, itself skipped for depending on "
|
||||
"unavailable module %s.%s",
|
||||
module.__name__,
|
||||
dependency.__name__,
|
||||
self._dependency_name(declared, dependency),
|
||||
missing.__name__,
|
||||
remainder,
|
||||
)
|
||||
@@ -383,7 +631,7 @@ class Command:
|
||||
"module %s, which depends on unavailable "
|
||||
"module %s.%s",
|
||||
module.__name__,
|
||||
dependency.__name__,
|
||||
self._dependency_name(declared, dependency),
|
||||
root.__name__,
|
||||
missing.__name__,
|
||||
remainder,
|
||||
@@ -397,12 +645,9 @@ class Command:
|
||||
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]
|
||||
selected = self._selected_modules(modules)
|
||||
if selected is None:
|
||||
return None
|
||||
|
||||
required: set[type[MVTModule]] = set()
|
||||
pending = list(selected)
|
||||
@@ -411,7 +656,7 @@ class Command:
|
||||
if module in required:
|
||||
continue
|
||||
required.add(module)
|
||||
for dependency in module.dependencies:
|
||||
for _, dependency in self._module_dependencies(module):
|
||||
# Unavailable dependencies are reported by _skipped_modules().
|
||||
if dependency in module_indexes:
|
||||
pending.append(dependency)
|
||||
@@ -424,13 +669,14 @@ class Command:
|
||||
"Every selected module was skipped for an unavailable "
|
||||
"dependency. No modules will be run."
|
||||
)
|
||||
return None
|
||||
|
||||
dependents: dict[type[MVTModule], list[type[MVTModule]]] = {
|
||||
module: [] for module in runnable
|
||||
}
|
||||
indegree = {module: 0 for module in runnable}
|
||||
for module in runnable:
|
||||
for dependency in module.dependencies:
|
||||
for _, dependency in self._module_dependencies(module):
|
||||
if dependency not in indegree:
|
||||
continue
|
||||
dependents[dependency].append(module)
|
||||
@@ -486,9 +732,12 @@ class Command:
|
||||
module_options=self.module_options,
|
||||
log=module_logger,
|
||||
)
|
||||
# Dependencies are keyed by the module class they declare, even
|
||||
# when it was replaced, so that a module asking for the results
|
||||
# of a replaced module receives those of its replacement.
|
||||
m.dependency_modules = {
|
||||
dependency: executed_by_type[dependency]
|
||||
for dependency in module.dependencies
|
||||
dependency: executed_by_type[resolved]
|
||||
for dependency, resolved in self._module_dependencies(module)
|
||||
}
|
||||
|
||||
if self.iocs.total_ioc_count:
|
||||
|
||||
@@ -17,6 +17,10 @@ HELP_MSG_LOAD_MODULE = (
|
||||
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"
|
||||
HELP_MSG_VERBOSE_COMMAND = (
|
||||
"Verbose mode (kept for compatibility, pass --verbose before the command "
|
||||
"name instead)"
|
||||
)
|
||||
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"
|
||||
|
||||
@@ -46,6 +46,11 @@ class MVTModule:
|
||||
slug: Optional[str] = None
|
||||
dependencies: Sequence[type["MVTModule"]] = ()
|
||||
supported_commands: Sequence[tuple[str, str]] = ()
|
||||
# A custom module can name a module class it supersedes, usually a
|
||||
# built-in one. When both are available to a command, the named module is
|
||||
# dropped from the run and this module takes its place, including in the
|
||||
# dependencies of any other module.
|
||||
replaces: Optional[type["MVTModule"]] = None
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
|
||||
@@ -10,6 +10,7 @@ import inspect
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
from functools import lru_cache
|
||||
@@ -26,6 +27,9 @@ EXTERNAL_LOGGER_NAMESPACE = "mvt.ext"
|
||||
PLUGIN_PACKAGE_PREFIX = "mvt_plugin_"
|
||||
_ORIGIN_ATTRIBUTE = "_mvt_module_origin"
|
||||
_PATH_MODULE_PREFIX = "_mvt_custom_module_"
|
||||
# Shared with cli_plugins, which names a loaded command file this way.
|
||||
CUSTOM_COMMAND_MODULE_PREFIX = "_mvt_custom_command_"
|
||||
_LOADED_FILE_DIGEST = re.compile(r"_[0-9a-f]{16}$")
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@@ -66,6 +70,25 @@ def _module_name_for_path(path: Path) -> str:
|
||||
return f"{_PATH_MODULE_PREFIX}{path.stem}_{digest}"
|
||||
|
||||
|
||||
def _is_builtin_logger_name(name: str) -> bool:
|
||||
return name == "mvt" or name.startswith("mvt.")
|
||||
|
||||
|
||||
def _loaded_file_stem(name: str, prefix: str) -> str:
|
||||
"""Recover a loaded file's name from the import name MVT gave it."""
|
||||
return _LOADED_FILE_DIGEST.sub("", name[len(prefix) :])
|
||||
|
||||
|
||||
def _external_logger_name(name: str) -> str:
|
||||
"""Return the "mvt.ext" logger name external code logs under."""
|
||||
top_level, separator, rest = name.partition(".")
|
||||
if top_level.startswith(PLUGIN_PACKAGE_PREFIX) and len(top_level) > len(
|
||||
PLUGIN_PACKAGE_PREFIX
|
||||
):
|
||||
name = top_level[len(PLUGIN_PACKAGE_PREFIX) :] + separator + rest
|
||||
return f"{EXTERNAL_LOGGER_NAMESPACE}.{name}"
|
||||
|
||||
|
||||
def get_module_logger(module_class: type[MVTModule]) -> logging.Logger:
|
||||
"""Return the logger a module's records should be emitted through.
|
||||
|
||||
@@ -80,19 +103,35 @@ def get_module_logger(module_class: type[MVTModule]) -> logging.Logger:
|
||||
"mvt_plugin_<name>" naming convention log under "mvt.ext.<name>".
|
||||
"""
|
||||
name = module_class.__module__
|
||||
if name == "mvt" or name.startswith("mvt."):
|
||||
if _is_builtin_logger_name(name):
|
||||
return logging.getLogger(name)
|
||||
|
||||
if name.startswith(_PATH_MODULE_PREFIX):
|
||||
name = Path(get_module_origin(module_class).name).stem
|
||||
else:
|
||||
top_level, separator, rest = name.partition(".")
|
||||
if top_level.startswith(PLUGIN_PACKAGE_PREFIX) and len(top_level) > len(
|
||||
PLUGIN_PACKAGE_PREFIX
|
||||
):
|
||||
name = top_level[len(PLUGIN_PACKAGE_PREFIX) :] + separator + rest
|
||||
file_name = Path(get_module_origin(module_class).name).stem
|
||||
return logging.getLogger(f"{EXTERNAL_LOGGER_NAMESPACE}.{file_name}")
|
||||
|
||||
return logging.getLogger(f"{EXTERNAL_LOGGER_NAMESPACE}.{name}")
|
||||
return logging.getLogger(_external_logger_name(name))
|
||||
|
||||
|
||||
def get_plugin_logger(name: str) -> logging.Logger:
|
||||
"""Return a general logger for use in custom MVT plugins.
|
||||
|
||||
Call it with ``__name__``. The logger sits under "mvt.ext". That is
|
||||
where get_module_logger() puts module classes. A file loaded with
|
||||
--load-module or --load-command is named after the file.
|
||||
"""
|
||||
if _is_builtin_logger_name(name):
|
||||
return logging.getLogger(name)
|
||||
|
||||
# A file loaded with --load-module or --load-command is imported under a
|
||||
# mangled name. Log it under the file it came from. get_module_logger()
|
||||
# does the same for the module classes such a file defines.
|
||||
for prefix in (_PATH_MODULE_PREFIX, CUSTOM_COMMAND_MODULE_PREFIX):
|
||||
if name.startswith(prefix):
|
||||
stem = _loaded_file_stem(name, prefix)
|
||||
return logging.getLogger(f"{EXTERNAL_LOGGER_NAMESPACE}.{stem}")
|
||||
|
||||
return logging.getLogger(_external_logger_name(name))
|
||||
|
||||
|
||||
def _iter_module_files(path: Path) -> Iterable[Path]:
|
||||
|
||||
+11
-5
@@ -256,12 +256,18 @@ def init_logging(verbose: bool = False):
|
||||
|
||||
|
||||
def set_verbose_logging(verbose: bool = False):
|
||||
"""Raise or lower the verbosity of MVT's console output.
|
||||
|
||||
Only MVT's own console handler is adjusted, wherever it sits in the list.
|
||||
The file handler a command attaches to its output folder keeps recording
|
||||
everything, so the command.log of a run does not depend on how the run was
|
||||
invoked, and a handler attached to the "mvt" logger by anything else is
|
||||
left alone.
|
||||
"""
|
||||
log = logging.getLogger("mvt")
|
||||
handler = log.handlers[0]
|
||||
if verbose:
|
||||
handler.setLevel(logging.DEBUG)
|
||||
else:
|
||||
handler.setLevel(logging.INFO)
|
||||
for handler in log.handlers:
|
||||
if isinstance(handler, MVTLogHandler):
|
||||
handler.setLevel(logging.DEBUG if verbose else logging.INFO)
|
||||
|
||||
|
||||
def exec_or_profile(module, globals, locals):
|
||||
|
||||
+18
-11
@@ -38,6 +38,7 @@ from mvt.common.help import (
|
||||
HELP_MSG_LOAD_MODULE,
|
||||
HELP_MSG_MODULE,
|
||||
HELP_MSG_VERBOSE,
|
||||
HELP_MSG_VERBOSE_COMMAND,
|
||||
HELP_MSG_CHECK_FS,
|
||||
HELP_MSG_CHECK_IOCS,
|
||||
HELP_MSG_STIX2,
|
||||
@@ -52,9 +53,7 @@ from .cmd_check_backup import CmdIOSCheckBackup
|
||||
from .cmd_check_fs import CmdIOSCheckFS
|
||||
from .cmd_check_sysdiagnose import CmdIOSCheckSysdiagnose
|
||||
from .decrypt import DecryptBackup
|
||||
from .modules.backup import BACKUP_MODULES
|
||||
from .modules.fs import FS_MODULES
|
||||
from .modules.mixed import MIXED_MODULES
|
||||
from .command_modules import IOS_CHECK_IOCS_MODULES
|
||||
|
||||
init_logging()
|
||||
log = logging.getLogger("mvt")
|
||||
@@ -74,6 +73,11 @@ def _get_disable_flags(ctx):
|
||||
)
|
||||
|
||||
|
||||
def _get_verbose(ctx):
|
||||
"""Return whether --verbose was passed to the CLI itself."""
|
||||
return bool(ctx.obj and ctx.obj.get("verbose", False))
|
||||
|
||||
|
||||
def _load_custom_modules(load_module):
|
||||
try:
|
||||
return load_custom_modules(load_module)
|
||||
@@ -94,11 +98,14 @@ def _load_custom_modules(load_module):
|
||||
is_flag=True,
|
||||
help=HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK,
|
||||
)
|
||||
@click.option("--verbose", "-v", is_flag=True, help=HELP_MSG_VERBOSE)
|
||||
@click.pass_context
|
||||
def cli(ctx, disable_update_check, disable_indicator_update_check):
|
||||
def cli(ctx, disable_update_check, disable_indicator_update_check, verbose):
|
||||
ctx.ensure_object(dict)
|
||||
ctx.obj["disable_version_check"] = disable_update_check
|
||||
ctx.obj["disable_indicator_check"] = disable_indicator_update_check
|
||||
ctx.obj["verbose"] = verbose
|
||||
set_verbose_logging(verbose)
|
||||
logo(
|
||||
disable_version_check=disable_update_check,
|
||||
disable_indicator_check=disable_indicator_update_check,
|
||||
@@ -254,7 +261,7 @@ def extract_key(password, key_file, backup_path):
|
||||
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.option("--verbose", "-v", is_flag=True, help=HELP_MSG_VERBOSE_COMMAND)
|
||||
@click.argument("BACKUP_PATH", type=click.Path(exists=True))
|
||||
@click.pass_context
|
||||
def check_backup(
|
||||
@@ -269,7 +276,7 @@ def check_backup(
|
||||
verbose,
|
||||
backup_path,
|
||||
):
|
||||
set_verbose_logging(verbose)
|
||||
set_verbose_logging(verbose or _get_verbose(ctx))
|
||||
module_options = {"fast_mode": fast}
|
||||
custom_modules = _load_custom_modules(load_module)
|
||||
|
||||
@@ -323,7 +330,7 @@ def check_backup(
|
||||
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.option("--verbose", "-v", is_flag=True, help=HELP_MSG_VERBOSE_COMMAND)
|
||||
@click.argument("DUMP_PATH", type=click.Path(exists=True))
|
||||
@click.pass_context
|
||||
def check_fs(
|
||||
@@ -338,7 +345,7 @@ def check_fs(
|
||||
verbose,
|
||||
dump_path,
|
||||
):
|
||||
set_verbose_logging(verbose)
|
||||
set_verbose_logging(verbose or _get_verbose(ctx))
|
||||
module_options = {"fast_mode": fast}
|
||||
custom_modules = _load_custom_modules(load_module)
|
||||
|
||||
@@ -392,7 +399,7 @@ def check_fs(
|
||||
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.option("--verbose", "-v", is_flag=True, help=HELP_MSG_VERBOSE_COMMAND)
|
||||
@click.argument("SYSDIAGNOSE_PATH", type=click.Path(exists=True))
|
||||
@click.pass_context
|
||||
def check_sysdiagnose(
|
||||
@@ -406,7 +413,7 @@ def check_sysdiagnose(
|
||||
verbose,
|
||||
sysdiagnose_path,
|
||||
):
|
||||
set_verbose_logging(verbose)
|
||||
set_verbose_logging(verbose or _get_verbose(ctx))
|
||||
custom_modules = _load_custom_modules(load_module)
|
||||
cmd = CmdIOSCheckSysdiagnose(
|
||||
target_path=sysdiagnose_path,
|
||||
@@ -470,7 +477,7 @@ def check_iocs(ctx, iocs, list_modules, module, load_module, folder):
|
||||
custom_modules=custom_modules,
|
||||
platform="ios",
|
||||
)
|
||||
cmd.modules = BACKUP_MODULES + FS_MODULES + MIXED_MODULES
|
||||
cmd.modules = IOS_CHECK_IOCS_MODULES
|
||||
|
||||
if list_modules:
|
||||
cmd.list_modules()
|
||||
|
||||
@@ -0,0 +1,22 @@
|
||||
# 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/
|
||||
|
||||
"""Module lists an mvt-ios command composes from more than one family.
|
||||
|
||||
Commands whose modules are one family read that family directly. check-iocs
|
||||
re-checks stored results, so it has to know every module that could have
|
||||
written one, and both the CLI and any other code needing that answer share
|
||||
the list from here rather than each concatenating their own.
|
||||
"""
|
||||
|
||||
from mvt.common.module import MVTModule
|
||||
|
||||
from .modules.backup import BACKUP_MODULES
|
||||
from .modules.fs import FS_MODULES
|
||||
from .modules.mixed import MIXED_MODULES
|
||||
|
||||
IOS_CHECK_IOCS_MODULES: list[type[MVTModule]] = (
|
||||
BACKUP_MODULES + FS_MODULES + MIXED_MODULES
|
||||
)
|
||||
@@ -0,0 +1,80 @@
|
||||
# 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/
|
||||
|
||||
"""Stable functions and modules which plugins can import.
|
||||
|
||||
Anything else in mvt can be imported too, and may change
|
||||
between releases without notice.
|
||||
"""
|
||||
|
||||
from mvt.android.modules.androidqf.base import AndroidQFModule
|
||||
from mvt.android.modules.backup.base import BackupModule as AndroidBackupModule
|
||||
from mvt.android.modules.bugreport.base import BugReportModule
|
||||
from mvt.common.alerts import Alert, AlertLevel
|
||||
from mvt.common.command import Command
|
||||
from mvt.common.config import settings
|
||||
from mvt.common.module import DatabaseCorruptedError, DatabaseNotFoundError, MVTModule
|
||||
from mvt.common.module_loader import get_plugin_logger
|
||||
from mvt.common.module_types import (
|
||||
ModuleAtomicResult,
|
||||
ModuleResults,
|
||||
ModuleSerializedResult,
|
||||
)
|
||||
from mvt.common.plugin_config import (
|
||||
MVTPluginSettings,
|
||||
PluginConfigLoadError,
|
||||
plugin_config_path,
|
||||
plugin_data_folder,
|
||||
plugin_env_prefix,
|
||||
)
|
||||
from mvt.common.utils import (
|
||||
convert_chrometime_to_datetime,
|
||||
convert_datetime_to_iso,
|
||||
convert_mactime_to_datetime,
|
||||
convert_mactime_to_iso,
|
||||
convert_unix_to_iso,
|
||||
convert_unix_to_utc_datetime,
|
||||
)
|
||||
from mvt.common.version import MVT_VERSION
|
||||
from mvt.ios.modules.base import IOSExtraction
|
||||
from mvt.ios.modules.sysdiagnose.base import SysdiagnoseExtraction
|
||||
|
||||
__all__ = [
|
||||
# Classes a plugin subclasses.
|
||||
"MVTModule",
|
||||
"IOSExtraction",
|
||||
"SysdiagnoseExtraction",
|
||||
"AndroidQFModule",
|
||||
"AndroidBackupModule",
|
||||
"BugReportModule",
|
||||
"Command",
|
||||
# Results and alerts.
|
||||
"ModuleAtomicResult",
|
||||
"ModuleResults",
|
||||
"ModuleSerializedResult",
|
||||
"Alert",
|
||||
"AlertLevel",
|
||||
# Errors a module raises.
|
||||
"DatabaseNotFoundError",
|
||||
"DatabaseCorruptedError",
|
||||
# Settings.
|
||||
"settings",
|
||||
"MVTPluginSettings",
|
||||
"PluginConfigLoadError",
|
||||
"plugin_config_path",
|
||||
"plugin_data_folder",
|
||||
"plugin_env_prefix",
|
||||
# Logging.
|
||||
"get_plugin_logger",
|
||||
# Timestamps.
|
||||
"convert_chrometime_to_datetime",
|
||||
"convert_datetime_to_iso",
|
||||
"convert_mactime_to_datetime",
|
||||
"convert_mactime_to_iso",
|
||||
"convert_unix_to_iso",
|
||||
"convert_unix_to_utc_datetime",
|
||||
# MVT's version.
|
||||
"MVT_VERSION",
|
||||
]
|
||||
@@ -0,0 +1,139 @@
|
||||
# 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 json
|
||||
import logging
|
||||
|
||||
import pytest
|
||||
from click.testing import CliRunner
|
||||
|
||||
from mvt.android.cli import cli as android_cli
|
||||
from mvt.android.command_modules import ANDROID_CHECK_IOCS_MODULES
|
||||
from mvt.common.cmd_check_iocs import CmdCheckIOCS
|
||||
from mvt.common.module import MVTModule
|
||||
from mvt.ios.cli import cli as ios_cli
|
||||
from mvt.ios.command_modules import IOS_CHECK_IOCS_MODULES
|
||||
|
||||
# Keep the banner of the group callback from checking for updates online.
|
||||
OFFLINE = ["--disable-update-check", "--disable-indicator-update-check"]
|
||||
|
||||
|
||||
class CustomResultsModule(MVTModule):
|
||||
"""A custom module which declares the check-iocs pair of both platforms."""
|
||||
|
||||
slug = "custom_results"
|
||||
supported_commands = (
|
||||
("ios", "check-backup"),
|
||||
("ios", "check-iocs"),
|
||||
("android", "check-iocs"),
|
||||
)
|
||||
|
||||
checked: list = []
|
||||
|
||||
def run(self) -> None:
|
||||
pass
|
||||
|
||||
def check_indicators(self) -> None:
|
||||
self.checked.append(list(self.results))
|
||||
|
||||
|
||||
class BackupOnlyModule(MVTModule):
|
||||
"""A custom module which does not declare check-iocs."""
|
||||
|
||||
slug = "backup_only"
|
||||
supported_commands = (("ios", "check-backup"),)
|
||||
|
||||
def check_indicators(self) -> None:
|
||||
raise AssertionError("must not be re-checked")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"platform, builtin_modules",
|
||||
[("ios", IOS_CHECK_IOCS_MODULES), ("android", ANDROID_CHECK_IOCS_MODULES)],
|
||||
)
|
||||
def test_check_iocs_rechecks_the_stored_results_of_custom_modules(
|
||||
platform, builtin_modules, tmp_path, caplog
|
||||
):
|
||||
# check-iocs matches every <slug>.json in the results folder to the module
|
||||
# with that slug, custom modules included, and runs its check_indicators()
|
||||
# again over the stored results.
|
||||
results = [{"domain": "example.org"}]
|
||||
(tmp_path / "custom_results.json").write_text(json.dumps(results))
|
||||
(tmp_path / "backup_only.json").write_text(json.dumps(results))
|
||||
CustomResultsModule.checked.clear()
|
||||
|
||||
cmd = CmdCheckIOCS(
|
||||
target_path=str(tmp_path),
|
||||
custom_modules=[CustomResultsModule, BackupOnlyModule],
|
||||
platform=platform,
|
||||
)
|
||||
cmd.modules = builtin_modules
|
||||
|
||||
with caplog.at_level(logging.INFO):
|
||||
cmd.run()
|
||||
|
||||
assert CustomResultsModule.checked == [results]
|
||||
assert (
|
||||
'Loading results from "custom_results.json" with module CustomResultsModule'
|
||||
in caplog.text
|
||||
)
|
||||
# A module declaring only check-backup is not part of check-iocs.
|
||||
assert "backup_only.json" not in caplog.text
|
||||
|
||||
|
||||
def test_check_iocs_lists_custom_modules_declaring_the_command(caplog):
|
||||
cmd = CmdCheckIOCS(
|
||||
custom_modules=[CustomResultsModule, BackupOnlyModule],
|
||||
platform="ios",
|
||||
)
|
||||
cmd.modules = IOS_CHECK_IOCS_MODULES
|
||||
|
||||
with caplog.at_level(logging.INFO):
|
||||
cmd.list_modules()
|
||||
|
||||
assert "CustomResultsModule" in caplog.text
|
||||
assert "BackupOnlyModule" not in caplog.text
|
||||
|
||||
|
||||
LOADED_MODULE = '''
|
||||
from mvt.common.module import MVTModule
|
||||
|
||||
|
||||
class LoadedResultsModule(MVTModule):
|
||||
"""A module loaded from a file with --load-module."""
|
||||
|
||||
slug = "loaded_results"
|
||||
supported_commands = (("ios", "check-iocs"), ("android", "check-iocs"))
|
||||
|
||||
def run(self) -> None:
|
||||
pass
|
||||
|
||||
def check_indicators(self) -> None:
|
||||
self.log.warning("loaded module checked %d results", len(self.results))
|
||||
'''
|
||||
|
||||
|
||||
@pytest.mark.parametrize("cli", [ios_cli, android_cli], ids=["mvt-ios", "mvt-android"])
|
||||
def test_check_iocs_loads_custom_modules_from_a_file_on_each_cli(cli, tmp_path, caplog):
|
||||
module_path = tmp_path / "loaded_module.py"
|
||||
module_path.write_text(LOADED_MODULE)
|
||||
results_folder = tmp_path / "results"
|
||||
results_folder.mkdir()
|
||||
(results_folder / "loaded_results.json").write_text(json.dumps([{"a": 1}]))
|
||||
|
||||
with caplog.at_level(logging.INFO):
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
[
|
||||
*OFFLINE,
|
||||
"check-iocs",
|
||||
"--load-module",
|
||||
str(module_path),
|
||||
str(results_folder),
|
||||
],
|
||||
)
|
||||
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "loaded module checked 1 results" in caplog.text
|
||||
@@ -68,6 +68,123 @@ class CustomDependsOnBuiltin(RecordingModule):
|
||||
dependencies = (FirstModule,)
|
||||
|
||||
|
||||
class ReplacementModule(FirstModule):
|
||||
supported_commands = (("ios", "check-backup"),)
|
||||
replaces = FirstModule
|
||||
|
||||
def run(self):
|
||||
super().run()
|
||||
self.results = ["replacement"]
|
||||
|
||||
|
||||
class OtherReplacementModule(FirstModule):
|
||||
supported_commands = (("ios", "check-backup"),)
|
||||
replaces = FirstModule
|
||||
|
||||
|
||||
class UnrelatedReplacementModule(RecordingModule):
|
||||
supported_commands = (("ios", "check-backup"),)
|
||||
replaces = IndependentModule
|
||||
|
||||
|
||||
class ReplacementOfReplacementModule(ReplacementModule):
|
||||
supported_commands = (("ios", "check-backup"),)
|
||||
replaces = ReplacementModule
|
||||
|
||||
def run(self):
|
||||
super().run()
|
||||
self.results = ["replacement of replacement"]
|
||||
|
||||
|
||||
class ReplacesOwnDependencyModule(FirstModule):
|
||||
supported_commands = (("ios", "check-backup"),)
|
||||
replaces = FirstModule
|
||||
dependencies = (FirstModule,)
|
||||
|
||||
|
||||
class DisabledReplacementModule(FirstModule):
|
||||
supported_commands = (("ios", "check-backup"),)
|
||||
replaces = FirstModule
|
||||
enabled = False
|
||||
|
||||
|
||||
class MutualReplacementOne(RecordingModule):
|
||||
supported_commands = (("ios", "check-backup"),)
|
||||
|
||||
|
||||
class MutualReplacementTwo(RecordingModule):
|
||||
supported_commands = (("ios", "check-backup"),)
|
||||
replaces = MutualReplacementOne
|
||||
|
||||
|
||||
MutualReplacementOne.replaces = MutualReplacementTwo
|
||||
|
||||
|
||||
class CycleOneModule(RecordingModule):
|
||||
supported_commands = (("ios", "check-backup"),)
|
||||
|
||||
|
||||
class CycleTwoModule(RecordingModule):
|
||||
supported_commands = (("ios", "check-backup"),)
|
||||
replaces = CycleOneModule
|
||||
|
||||
|
||||
class CycleThreeModule(RecordingModule):
|
||||
supported_commands = (("ios", "check-backup"),)
|
||||
replaces = CycleTwoModule
|
||||
|
||||
|
||||
CycleOneModule.replaces = CycleThreeModule
|
||||
|
||||
|
||||
class UnavailableDependencyModule(RecordingModule):
|
||||
supported_commands = (("ios", "check-backup"),)
|
||||
|
||||
|
||||
class ReplacementMissingDependency(FirstModule):
|
||||
supported_commands = (("ios", "check-backup"),)
|
||||
replaces = FirstModule
|
||||
dependencies = (UnavailableDependencyModule,)
|
||||
|
||||
|
||||
class SharedSlugModule(RecordingModule):
|
||||
supported_commands = (("ios", "check-backup"),)
|
||||
slug = "shared_slug"
|
||||
|
||||
|
||||
class OtherSharedSlugModule(RecordingModule):
|
||||
supported_commands = (("ios", "check-backup"),)
|
||||
slug = "shared_slug"
|
||||
|
||||
|
||||
class ReplacementKeepingTheSlug(FirstModule):
|
||||
supported_commands = (("ios", "check-backup"),)
|
||||
replaces = FirstModule
|
||||
slug = "first_module"
|
||||
|
||||
|
||||
class SameNameReplacementModule(FirstModule):
|
||||
supported_commands = (("ios", "check-backup"),)
|
||||
replaces = FirstModule
|
||||
|
||||
def run(self):
|
||||
super().run()
|
||||
self.results = ["replacement"]
|
||||
|
||||
|
||||
# Modules replacing a built-in one usually keep its class name, which is the
|
||||
# name `--module` matches on.
|
||||
SameNameReplacementModule.__name__ = "FirstModule"
|
||||
|
||||
|
||||
def logged_substitutions(caplog) -> list[str]:
|
||||
return [
|
||||
record.getMessage()
|
||||
for record in caplog.records
|
||||
if record.levelno == logging.INFO and "replaces module" in record.getMessage()
|
||||
]
|
||||
|
||||
|
||||
class RecordingCommand(Command):
|
||||
def init(self):
|
||||
self.initialized = True
|
||||
@@ -303,3 +420,376 @@ class TestCommand:
|
||||
cmd.run()
|
||||
|
||||
assert RecordingModule.run_order == ["FirstModule", "CustomDependsOnBuiltin"]
|
||||
|
||||
def test_custom_module_replaces_builtin(self, caplog):
|
||||
cmd = RecordingCommand()
|
||||
cmd.platform = "ios"
|
||||
cmd.name = "check-backup"
|
||||
cmd.modules = [FirstModule, IndependentModule]
|
||||
cmd.custom_modules = [ReplacementModule]
|
||||
|
||||
with caplog.at_level(logging.INFO):
|
||||
cmd.run()
|
||||
|
||||
assert RecordingModule.run_order == ["IndependentModule", "ReplacementModule"]
|
||||
assert cmd.module_replacements == {FirstModule: ReplacementModule}
|
||||
assert (
|
||||
"Module ReplacementModule from" in caplog.text
|
||||
and "replaces module FirstModule" in caplog.text
|
||||
)
|
||||
|
||||
def test_modules_without_replaces_all_run(self):
|
||||
cmd = RecordingCommand()
|
||||
cmd.platform = "ios"
|
||||
cmd.name = "check-backup"
|
||||
cmd.modules = [FirstModule]
|
||||
cmd.custom_modules = [CustomIOSBackupModule]
|
||||
|
||||
cmd.run()
|
||||
|
||||
assert RecordingModule.run_order == ["FirstModule", "CustomIOSBackupModule"]
|
||||
assert cmd.module_replacements == {}
|
||||
|
||||
def test_unavailable_replaced_module_is_ignored(self, caplog):
|
||||
cmd = RecordingCommand()
|
||||
cmd.platform = "ios"
|
||||
cmd.name = "check-backup"
|
||||
cmd.modules = [IndependentModule]
|
||||
cmd.custom_modules = [ReplacementModule]
|
||||
|
||||
with caplog.at_level(logging.INFO):
|
||||
cmd.run()
|
||||
|
||||
assert RecordingModule.run_order == ["IndependentModule", "ReplacementModule"]
|
||||
assert cmd.module_replacements == {}
|
||||
assert "replaces module FirstModule" not in caplog.text
|
||||
|
||||
def test_dependencies_are_resolved_to_the_replacement(self):
|
||||
cmd = RecordingCommand()
|
||||
cmd.platform = "ios"
|
||||
cmd.name = "check-backup"
|
||||
cmd.modules = [SecondModule, FirstModule]
|
||||
cmd.custom_modules = [ReplacementModule]
|
||||
|
||||
cmd.run()
|
||||
|
||||
assert RecordingModule.run_order == ["ReplacementModule", "SecondModule"]
|
||||
second = next(
|
||||
module for module in cmd.executed if isinstance(module, SecondModule)
|
||||
)
|
||||
assert isinstance(second.dependency_modules[FirstModule], ReplacementModule)
|
||||
assert second.results == ["replacement", "second"]
|
||||
|
||||
def test_replacement_with_unavailable_dependency_skips_its_dependents(
|
||||
self, caplog
|
||||
):
|
||||
cmd = RecordingCommand()
|
||||
cmd.platform = "ios"
|
||||
cmd.name = "check-backup"
|
||||
cmd.modules = [FirstModule, SecondModule, IndependentModule]
|
||||
cmd.custom_modules = [ReplacementMissingDependency]
|
||||
|
||||
with caplog.at_level(logging.INFO):
|
||||
cmd.run()
|
||||
|
||||
# The replacement cannot run, and the module it replaced was dropped
|
||||
# from the run by the replacement, so neither of them produces
|
||||
# results and the module depending on the replaced one is skipped.
|
||||
assert RecordingModule.run_order == ["IndependentModule"]
|
||||
assert (
|
||||
"Module ReplacementMissingDependency will be SKIPPED: it depends "
|
||||
"on module UnavailableDependencyModule, which is not available in "
|
||||
"this command." in caplog.text
|
||||
)
|
||||
# The skipped dependent is told which module it actually depends on,
|
||||
# and the module class its author declared.
|
||||
assert (
|
||||
"Module SecondModule will be SKIPPED: it depends on module "
|
||||
"ReplacementMissingDependency (replacing module FirstModule), "
|
||||
"itself skipped for depending on unavailable module "
|
||||
"UnavailableDependencyModule." in caplog.text
|
||||
)
|
||||
|
||||
def test_selected_replacement_with_unavailable_dependency_runs_nothing(
|
||||
self, caplog, tmp_path
|
||||
):
|
||||
cmd = RecordingCommand(module_name="FirstModule", results_path=str(tmp_path))
|
||||
cmd.platform = "ios"
|
||||
cmd.name = "check-backup"
|
||||
cmd.modules = [FirstModule, IndependentModule]
|
||||
cmd.custom_modules = [ReplacementMissingDependency]
|
||||
|
||||
with caplog.at_level(logging.INFO):
|
||||
cmd.run()
|
||||
|
||||
assert RecordingModule.run_order == []
|
||||
assert not hasattr(cmd, "initialized")
|
||||
assert (
|
||||
"Module FirstModule was replaced by module "
|
||||
"ReplacementMissingDependency, which is run in its place."
|
||||
in caplog.text
|
||||
)
|
||||
assert "Module ReplacementMissingDependency will be SKIPPED" in caplog.text
|
||||
assert "No modules will be run" in caplog.text
|
||||
# Nothing else was selected, so the warnings must not promise that the
|
||||
# analysis continues right before saying that it does not.
|
||||
assert "The rest of the analysis will still run" not in caplog.text
|
||||
# No module ran, so no results were stored next to the command log.
|
||||
assert [path.name for path in tmp_path.iterdir()] == ["command.log"]
|
||||
|
||||
def test_chained_replacements_are_resolved(self):
|
||||
cmd = RecordingCommand()
|
||||
cmd.platform = "ios"
|
||||
cmd.name = "check-backup"
|
||||
cmd.modules = [SecondModule, FirstModule]
|
||||
cmd.custom_modules = [ReplacementModule, ReplacementOfReplacementModule]
|
||||
|
||||
cmd.run()
|
||||
|
||||
assert RecordingModule.run_order == [
|
||||
"ReplacementOfReplacementModule",
|
||||
"SecondModule",
|
||||
]
|
||||
second = next(
|
||||
module for module in cmd.executed if isinstance(module, SecondModule)
|
||||
)
|
||||
assert second.results == ["replacement of replacement", "second"]
|
||||
|
||||
def test_replacement_which_is_not_a_subclass_warns(self, caplog):
|
||||
cmd = RecordingCommand()
|
||||
cmd.platform = "ios"
|
||||
cmd.name = "check-backup"
|
||||
cmd.modules = [IndependentModule]
|
||||
cmd.custom_modules = [UnrelatedReplacementModule]
|
||||
|
||||
with caplog.at_level(logging.WARNING):
|
||||
cmd.run()
|
||||
|
||||
assert RecordingModule.run_order == ["UnrelatedReplacementModule"]
|
||||
assert "is not a subclass of it" in caplog.text
|
||||
|
||||
def test_multiple_modules_replacing_the_same_module_warn(self, caplog):
|
||||
cmd = RecordingCommand()
|
||||
cmd.platform = "ios"
|
||||
cmd.name = "check-backup"
|
||||
cmd.modules = [FirstModule]
|
||||
cmd.custom_modules = [ReplacementModule, OtherReplacementModule]
|
||||
|
||||
with caplog.at_level(logging.WARNING):
|
||||
cmd.run()
|
||||
|
||||
assert RecordingModule.run_order == [
|
||||
"ReplacementModule",
|
||||
"OtherReplacementModule",
|
||||
]
|
||||
assert cmd.module_replacements == {FirstModule: ReplacementModule}
|
||||
assert (
|
||||
"Modules ReplacementModule and OtherReplacementModule both replace "
|
||||
"module FirstModule. Both of them will run, FirstModule will not, "
|
||||
"and modules depending on FirstModule will use the results of "
|
||||
"ReplacementModule." in caplog.text
|
||||
)
|
||||
assert "overwrite each other's results file" in caplog.text
|
||||
|
||||
def test_module_replacing_its_own_dependency_runs(self, caplog):
|
||||
cmd = RecordingCommand()
|
||||
cmd.platform = "ios"
|
||||
cmd.name = "check-backup"
|
||||
cmd.modules = [FirstModule]
|
||||
cmd.custom_modules = [ReplacesOwnDependencyModule]
|
||||
|
||||
with caplog.at_level(logging.WARNING):
|
||||
cmd.run()
|
||||
|
||||
assert RecordingModule.run_order == ["ReplacesOwnDependencyModule"]
|
||||
assert (
|
||||
"Module ReplacesOwnDependencyModule depends on module FirstModule, "
|
||||
"which it also replaces" in caplog.text
|
||||
)
|
||||
|
||||
def test_literal_self_dependency_is_still_circular(self, caplog):
|
||||
class SelfDependentModule(RecordingModule):
|
||||
pass
|
||||
|
||||
SelfDependentModule.dependencies = (SelfDependentModule,)
|
||||
|
||||
cmd = RecordingCommand()
|
||||
cmd.modules = [SelfDependentModule]
|
||||
|
||||
with caplog.at_level(logging.WARNING):
|
||||
cmd.run()
|
||||
|
||||
assert RecordingModule.run_order == []
|
||||
assert not hasattr(cmd, "initialized")
|
||||
assert "Circular module dependency detected" in caplog.text
|
||||
|
||||
def test_disabled_replacement_keeps_the_replaced_module(self, caplog):
|
||||
cmd = RecordingCommand()
|
||||
cmd.platform = "ios"
|
||||
cmd.name = "check-backup"
|
||||
cmd.modules = [FirstModule]
|
||||
cmd.custom_modules = [DisabledReplacementModule]
|
||||
|
||||
with caplog.at_level(logging.INFO):
|
||||
cmd.run()
|
||||
|
||||
assert RecordingModule.run_order == ["FirstModule"]
|
||||
assert cmd.module_replacements == {}
|
||||
assert logged_substitutions(caplog) == []
|
||||
|
||||
def test_modules_replacing_each_other_all_run(self, caplog):
|
||||
cmd = RecordingCommand()
|
||||
cmd.platform = "ios"
|
||||
cmd.name = "check-backup"
|
||||
cmd.custom_modules = [MutualReplacementOne, MutualReplacementTwo]
|
||||
|
||||
with caplog.at_level(logging.INFO):
|
||||
cmd.run()
|
||||
|
||||
assert RecordingModule.run_order == [
|
||||
"MutualReplacementOne",
|
||||
"MutualReplacementTwo",
|
||||
]
|
||||
assert cmd.module_replacements == {}
|
||||
assert (
|
||||
"Modules MutualReplacementOne, MutualReplacementTwo replace each "
|
||||
"other in a cycle" in caplog.text
|
||||
)
|
||||
assert logged_substitutions(caplog) == []
|
||||
|
||||
def test_replacement_cycle_of_three_modules_all_run(self, caplog):
|
||||
cmd = RecordingCommand()
|
||||
cmd.platform = "ios"
|
||||
cmd.name = "check-backup"
|
||||
cmd.custom_modules = [CycleOneModule, CycleTwoModule, CycleThreeModule]
|
||||
|
||||
with caplog.at_level(logging.INFO):
|
||||
cmd.run()
|
||||
|
||||
assert RecordingModule.run_order == [
|
||||
"CycleOneModule",
|
||||
"CycleTwoModule",
|
||||
"CycleThreeModule",
|
||||
]
|
||||
assert cmd.module_replacements == {}
|
||||
assert (
|
||||
"Modules CycleOneModule, CycleThreeModule, CycleTwoModule replace "
|
||||
"each other in a cycle" in caplog.text
|
||||
)
|
||||
assert logged_substitutions(caplog) == []
|
||||
|
||||
def test_selected_replaced_module_name_runs_the_replacement(self, caplog):
|
||||
cmd = RecordingCommand(module_name="FirstModule")
|
||||
cmd.platform = "ios"
|
||||
cmd.name = "check-backup"
|
||||
cmd.modules = [FirstModule]
|
||||
cmd.custom_modules = [ReplacementModule]
|
||||
|
||||
with caplog.at_level(logging.INFO):
|
||||
cmd.run()
|
||||
|
||||
assert RecordingModule.run_order == ["ReplacementModule"]
|
||||
assert (
|
||||
"Module FirstModule was replaced by module ReplacementModule"
|
||||
in caplog.text
|
||||
)
|
||||
|
||||
def test_unknown_selected_module_warns_and_stops(self, caplog):
|
||||
cmd = RecordingCommand(module_name="NoSuchModule")
|
||||
cmd.name = "check-backup"
|
||||
cmd.modules = [FirstModule]
|
||||
|
||||
with caplog.at_level(logging.WARNING):
|
||||
cmd.run()
|
||||
|
||||
assert RecordingModule.run_order == []
|
||||
assert not hasattr(cmd, "initialized")
|
||||
assert (
|
||||
"No module named NoSuchModule is available for the check-backup "
|
||||
"command" in caplog.text
|
||||
)
|
||||
|
||||
def test_selected_module_name_matches_the_replacement(self):
|
||||
cmd = RecordingCommand(module_name="FirstModule")
|
||||
cmd.platform = "ios"
|
||||
cmd.name = "check-backup"
|
||||
cmd.modules = [FirstModule]
|
||||
cmd.custom_modules = [SameNameReplacementModule]
|
||||
|
||||
cmd.run()
|
||||
|
||||
assert len(cmd.executed) == 1
|
||||
assert isinstance(cmd.executed[0], SameNameReplacementModule)
|
||||
assert cmd.executed[0].results == ["replacement"]
|
||||
|
||||
def test_list_modules_reflects_replacements(self, caplog):
|
||||
cmd = RecordingCommand()
|
||||
cmd.platform = "ios"
|
||||
cmd.name = "check-backup"
|
||||
cmd.modules = [FirstModule, IndependentModule]
|
||||
cmd.custom_modules = [ReplacementModule]
|
||||
|
||||
with caplog.at_level(logging.INFO):
|
||||
cmd.list_modules()
|
||||
|
||||
listed = [
|
||||
record.getMessage()
|
||||
for record in caplog.records
|
||||
if "Modules from" in record.getMessage()
|
||||
]
|
||||
assert any("ReplacementModule" in message for message in listed)
|
||||
assert not any("FirstModule" in message for message in listed)
|
||||
|
||||
def test_modules_sharing_a_slug_are_reported(self, caplog):
|
||||
cmd = RecordingCommand()
|
||||
cmd.platform = "ios"
|
||||
cmd.name = "check-backup"
|
||||
cmd.custom_modules = [SharedSlugModule, OtherSharedSlugModule]
|
||||
|
||||
with caplog.at_level(logging.WARNING):
|
||||
cmd.run()
|
||||
|
||||
assert RecordingModule.run_order == [
|
||||
"SharedSlugModule",
|
||||
"OtherSharedSlugModule",
|
||||
]
|
||||
collisions = [
|
||||
record.getMessage()
|
||||
for record in caplog.records
|
||||
if "both use the slug" in record.getMessage()
|
||||
]
|
||||
assert len(collisions) == 1
|
||||
assert "Modules SharedSlugModule from" in collisions[0]
|
||||
assert "and OtherSharedSlugModule from" in collisions[0]
|
||||
assert "both use the slug shared_slug" in collisions[0]
|
||||
assert "overwrites the results of the other in shared_slug.json" in (
|
||||
collisions[0]
|
||||
)
|
||||
|
||||
def test_replacement_keeping_the_replaced_slug_is_not_reported(self, caplog):
|
||||
cmd = RecordingCommand()
|
||||
cmd.platform = "ios"
|
||||
cmd.name = "check-backup"
|
||||
cmd.modules = [FirstModule]
|
||||
cmd.custom_modules = [ReplacementKeepingTheSlug]
|
||||
|
||||
with caplog.at_level(logging.WARNING):
|
||||
cmd.run()
|
||||
|
||||
# The replaced module is no longer part of the run, so taking over its
|
||||
# slug is what the replacement is for, not a collision.
|
||||
assert RecordingModule.run_order == ["ReplacementKeepingTheSlug"]
|
||||
assert ReplacementKeepingTheSlug.get_slug() == FirstModule.get_slug()
|
||||
assert "both use the slug" not in caplog.text
|
||||
|
||||
def test_modules_with_distinct_slugs_are_not_reported(self, caplog):
|
||||
cmd = RecordingCommand()
|
||||
cmd.platform = "ios"
|
||||
cmd.name = "check-backup"
|
||||
cmd.modules = [FirstModule, IndependentModule]
|
||||
cmd.custom_modules = [CustomIOSBackupModule]
|
||||
|
||||
with caplog.at_level(logging.WARNING):
|
||||
cmd.run()
|
||||
|
||||
assert "both use the slug" not in caplog.text
|
||||
@@ -0,0 +1,27 @@
|
||||
# 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/
|
||||
|
||||
from mvt.android.command_modules import ANDROID_CHECK_IOCS_MODULES
|
||||
from mvt.android.modules.androidqf import ANDROIDQF_MODULES
|
||||
from mvt.android.modules.backup import BACKUP_MODULES as ANDROID_BACKUP_MODULES
|
||||
from mvt.android.modules.bugreport import BUGREPORT_MODULES
|
||||
from mvt.android.modules.intrusion_logs import INTRUSION_LOGS_MODULES
|
||||
from mvt.ios.command_modules import IOS_CHECK_IOCS_MODULES
|
||||
from mvt.ios.modules.backup import BACKUP_MODULES as IOS_BACKUP_MODULES
|
||||
from mvt.ios.modules.fs import FS_MODULES
|
||||
from mvt.ios.modules.mixed import MIXED_MODULES
|
||||
|
||||
|
||||
def test_the_check_iocs_lists_are_the_families_of_their_platform():
|
||||
# The CLI reads these same lists, so nothing composing one elsewhere can
|
||||
# drift from what the command runs. This pins what the lists are composed
|
||||
# of.
|
||||
assert IOS_CHECK_IOCS_MODULES == IOS_BACKUP_MODULES + FS_MODULES + MIXED_MODULES
|
||||
assert ANDROID_CHECK_IOCS_MODULES == (
|
||||
ANDROID_BACKUP_MODULES
|
||||
+ BUGREPORT_MODULES
|
||||
+ ANDROIDQF_MODULES
|
||||
+ INTRUSION_LOGS_MODULES
|
||||
)
|
||||
@@ -1,9 +1,14 @@
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from mvt.common.cli_plugins import _module_name_for_path as _command_module_name
|
||||
from mvt.common.module import MVTModule
|
||||
from mvt.common.module_loader import (
|
||||
CustomModuleLoadError,
|
||||
_module_name_for_path,
|
||||
get_module_logger,
|
||||
get_plugin_logger,
|
||||
load_custom_modules,
|
||||
load_custom_modules_from_path,
|
||||
module_supports_command,
|
||||
@@ -168,9 +173,9 @@ def test_get_module_logger_strips_the_plugin_package_prefix():
|
||||
class PluginModule(MVTModule):
|
||||
pass
|
||||
|
||||
PluginModule.__module__ = "mvt_plugin_amnesty_custom.ios.custom"
|
||||
PluginModule.__module__ = "mvt_plugin_example_org.ios.custom"
|
||||
|
||||
assert get_module_logger(PluginModule).name == "mvt.ext.amnesty_custom.ios.custom"
|
||||
assert get_module_logger(PluginModule).name == "mvt.ext.example_org.ios.custom"
|
||||
|
||||
|
||||
def test_get_module_logger_only_strips_the_prefix_from_the_top_level():
|
||||
@@ -187,3 +192,25 @@ def test_get_module_logger_names_path_modules_after_their_file(tmp_path):
|
||||
module = load_custom_modules_from_path(str(module_path))[0]
|
||||
|
||||
assert get_module_logger(module).name == "mvt.ext.my_custom_module"
|
||||
|
||||
|
||||
def test_get_plugin_logger_uses_the_same_namespace_as_modules():
|
||||
assert (
|
||||
get_plugin_logger("mvt_plugin_example_org.commands.summarize").name
|
||||
== "mvt.ext.example_org.commands.summarize"
|
||||
)
|
||||
assert get_plugin_logger("example_plugin.cli").name == "mvt.ext.example_plugin.cli"
|
||||
|
||||
|
||||
def test_get_plugin_logger_keeps_builtin_names():
|
||||
assert get_plugin_logger("mvt.ios.cli").name == "mvt.ios.cli"
|
||||
|
||||
|
||||
def test_get_plugin_logger_names_loaded_files_after_the_file():
|
||||
# A file loaded with --load-command or --load-module is imported under a
|
||||
# mangled name. The log names the file instead.
|
||||
command_name = _command_module_name(Path("/tmp/case_summary.py"))
|
||||
module_name = _module_name_for_path(Path("/tmp/my_custom_module.py"))
|
||||
|
||||
assert get_plugin_logger(command_name).name == "mvt.ext.case_summary"
|
||||
assert get_plugin_logger(module_name).name == "mvt.ext.my_custom_module"
|
||||
@@ -0,0 +1,34 @@
|
||||
# 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 mvt.plugin
|
||||
from mvt.android.modules.backup.base import BackupModule
|
||||
from mvt.common.config import settings
|
||||
|
||||
from ..plugin_fixtures import run_isolated_python
|
||||
|
||||
|
||||
def test_the_exported_names_are_the_public_names():
|
||||
public = {name for name in vars(mvt.plugin) if not name.startswith("_")}
|
||||
|
||||
assert public == set(mvt.plugin.__all__)
|
||||
assert mvt.plugin.settings is settings
|
||||
assert mvt.plugin.AndroidBackupModule is BackupModule
|
||||
|
||||
|
||||
def test_the_surface_imports_before_anything_else_of_mvt(tmp_path):
|
||||
# A plugin can import the surface as its first import of MVT. The
|
||||
# subprocess gets a temporary home because importing MVT writes its
|
||||
# configuration file.
|
||||
result = run_isolated_python(
|
||||
"from mvt.plugin import IOSExtraction, MVT_VERSION, settings\n"
|
||||
"assert MVT_VERSION\n"
|
||||
"assert settings.NETWORK_TIMEOUT > 0\n"
|
||||
"assert IOSExtraction.__name__ == 'IOSExtraction'\n",
|
||||
home=tmp_path,
|
||||
)
|
||||
|
||||
assert result.returncode == 0, result.stderr
|
||||
assert result.stderr == ""
|
||||
@@ -18,6 +18,7 @@ from mvt.common.utils import (
|
||||
generate_hashes_from_path,
|
||||
get_sha256_from_file_path,
|
||||
init_logging,
|
||||
set_verbose_logging,
|
||||
)
|
||||
|
||||
from ..utils import get_artifact_folder
|
||||
@@ -122,3 +123,29 @@ class TestInitLogging:
|
||||
sum(isinstance(handler, MVTLogHandler) for handler in log.handlers)
|
||||
== handler_count
|
||||
)
|
||||
|
||||
def test_verbose_logging_finds_the_console_handler_among_others(self):
|
||||
# Something else may have attached a handler to the "mvt" logger
|
||||
# before MVT did, so the console handler is not always the first.
|
||||
log = logging.getLogger("mvt")
|
||||
init_logging()
|
||||
foreign_handler = logging.NullHandler()
|
||||
foreign_handler.setLevel(logging.CRITICAL)
|
||||
log.handlers.insert(0, foreign_handler)
|
||||
|
||||
try:
|
||||
set_verbose_logging(True)
|
||||
console_handlers = [
|
||||
handler
|
||||
for handler in log.handlers
|
||||
if isinstance(handler, MVTLogHandler)
|
||||
]
|
||||
assert console_handlers
|
||||
assert all(handler.level == logging.DEBUG for handler in console_handlers)
|
||||
assert foreign_handler.level == logging.CRITICAL
|
||||
|
||||
set_verbose_logging(False)
|
||||
assert all(handler.level == logging.INFO for handler in console_handlers)
|
||||
assert foreign_handler.level == logging.CRITICAL
|
||||
finally:
|
||||
log.handlers.remove(foreign_handler)
|
||||
@@ -73,6 +73,7 @@ def run_isolated_python(
|
||||
}
|
||||
isolated_environment["HOME"] = str(home)
|
||||
isolated_environment["XDG_CONFIG_HOME"] = str(home / "config")
|
||||
isolated_environment["XDG_DATA_HOME"] = str(home / "data")
|
||||
if site_path is not None:
|
||||
isolated_environment["PYTHONPATH"] = str(site_path)
|
||||
isolated_environment.update(environment)
|
||||
|
||||
@@ -0,0 +1,107 @@
|
||||
# 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 logging
|
||||
|
||||
import pytest
|
||||
from click.testing import CliRunner
|
||||
|
||||
from mvt.android.cli import cli as android_cli
|
||||
from mvt.cli import cli as mvt_cli
|
||||
from mvt.common.log import MVTLogHandler
|
||||
from mvt.common.utils import set_verbose_logging
|
||||
from mvt.ios.cli import cli as ios_cli
|
||||
|
||||
# Keep the banner of the group callback from checking for updates online.
|
||||
OFFLINE = ["--disable-update-check", "--disable-indicator-update-check"]
|
||||
|
||||
PROGRAMS = {"mvt": mvt_cli, "mvt-ios": ios_cli, "mvt-android": android_cli}
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _reset_console_level():
|
||||
"""Leave the console handler at its default level after every test."""
|
||||
yield
|
||||
set_verbose_logging(False)
|
||||
|
||||
|
||||
def _console_level():
|
||||
"""Return the level of MVT's own console log handler."""
|
||||
for handler in logging.getLogger("mvt").handlers:
|
||||
if isinstance(handler, MVTLogHandler):
|
||||
return handler.level
|
||||
raise AssertionError("MVT has no console log handler")
|
||||
|
||||
|
||||
class TestVerboseOnTheCommands:
|
||||
@pytest.mark.parametrize("program", sorted(PROGRAMS))
|
||||
def test_verbose_before_the_command_name_turns_on_debug(self, program):
|
||||
cli = PROGRAMS[program]
|
||||
|
||||
result = CliRunner().invoke(cli, [*OFFLINE, "--verbose", "version"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert _console_level() == logging.DEBUG
|
||||
|
||||
@pytest.mark.parametrize("program", sorted(PROGRAMS))
|
||||
def test_a_run_without_verbose_goes_back_to_info(self, program):
|
||||
cli = PROGRAMS[program]
|
||||
CliRunner().invoke(cli, [*OFFLINE, "--verbose", "version"])
|
||||
|
||||
result = CliRunner().invoke(cli, [*OFFLINE, "version"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert _console_level() == logging.INFO
|
||||
|
||||
def test_mvt_verbose_without_a_command_prints_the_help(self):
|
||||
result = CliRunner().invoke(mvt_cli, [*OFFLINE, "--verbose"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "Usage:" in result.output
|
||||
assert _console_level() == logging.DEBUG
|
||||
|
||||
|
||||
class TestVerboseOnTheCheckCommands:
|
||||
def test_ios_command_default_does_not_undo_the_cli_choice(self, tmp_path):
|
||||
result = CliRunner().invoke(
|
||||
ios_cli,
|
||||
[*OFFLINE, "--verbose", "check-backup", "--list-modules", str(tmp_path)],
|
||||
)
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert _console_level() == logging.DEBUG
|
||||
|
||||
def test_ios_verbose_after_the_command_name_still_works(self, tmp_path):
|
||||
result = CliRunner().invoke(
|
||||
ios_cli,
|
||||
[*OFFLINE, "check-backup", "--verbose", "--list-modules", str(tmp_path)],
|
||||
)
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert _console_level() == logging.DEBUG
|
||||
|
||||
def test_android_command_default_does_not_undo_the_cli_choice(self, tmp_path):
|
||||
result = CliRunner().invoke(
|
||||
android_cli,
|
||||
[*OFFLINE, "--verbose", "check-bugreport", "--list-modules", str(tmp_path)],
|
||||
)
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert _console_level() == logging.DEBUG
|
||||
|
||||
def test_android_verbose_after_the_command_name_still_works(self, tmp_path):
|
||||
result = CliRunner().invoke(
|
||||
android_cli,
|
||||
[*OFFLINE, "check-bugreport", "--verbose", "--list-modules", str(tmp_path)],
|
||||
)
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert _console_level() == logging.DEBUG
|
||||
|
||||
def test_the_command_option_says_it_is_kept_for_compatibility(self):
|
||||
result = CliRunner().invoke(ios_cli, [*OFFLINE, "check-backup", "--help"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "kept for compatibility" in result.output
|
||||
Reference in new issue
Block a user