mirror of
https://github.com/mvt-project/mvt.git
synced 2026-10-25 15:43:41 +01:00
Compare commits
7
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
efbab29f94 | ||
|
|
ecc22f54b7 | ||
|
|
90fc0ae351 | ||
|
|
dab82e38d0 | ||
|
|
4e031cd43f | ||
|
|
dac4acb180 | ||
|
|
dd8bd2cb01 |
No files matched your search
@@ -84,7 +84,7 @@ details.
|
|||||||
|
|
||||||
Users can also add top-level commands to `mvt-ios` and `mvt-android` from
|
Users can also add top-level commands to `mvt-ios` and `mvt-android` from
|
||||||
installed Python packages or local files and folders. See the
|
installed Python packages or local files and folders. See the
|
||||||
[custom CLI command documentation](https://docs.mvt.re/en/latest/custom_commands/)
|
[custom CLI command documentation](https://docs.mvt.re/en/latest/development/custom_commands/)
|
||||||
for the plugin entry points and `--load-command` interface.
|
for the plugin entry points and `--load-command` interface.
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -1,131 +0,0 @@
|
|||||||
# Development
|
|
||||||
|
|
||||||
The Mobile Verification Toolkit team welcomes contributions of new forensic modules or other contributions which help improve the software.
|
|
||||||
|
|
||||||
## Local environment
|
|
||||||
|
|
||||||
MVT uses `uv` for dependency management. To install the project and development dependencies from the locked environment, run:
|
|
||||||
|
|
||||||
```bash
|
|
||||||
make install
|
|
||||||
```
|
|
||||||
|
|
||||||
## Testing
|
|
||||||
|
|
||||||
MVT uses `pytest` for unit and integration tests. Code style consistency is maintained with `ruff` and `mypy`. All can
|
|
||||||
be run automatically with:
|
|
||||||
|
|
||||||
```bash
|
|
||||||
make check
|
|
||||||
```
|
|
||||||
|
|
||||||
Run these tests before making new commits or opening pull requests.
|
|
||||||
|
|
||||||
## Module dependencies
|
|
||||||
|
|
||||||
Modules can require other modules to run first by declaring their classes in
|
|
||||||
`dependencies`. The command runner uses a stable topological ordering, so the
|
|
||||||
existing module list order is preserved wherever dependency constraints allow.
|
|
||||||
|
|
||||||
```python
|
|
||||||
class DependentModule(MVTModule):
|
|
||||||
dependencies = (PrerequisiteModule,)
|
|
||||||
|
|
||||||
def run(self):
|
|
||||||
prerequisite_results = self.get_dependency_results(PrerequisiteModule)
|
|
||||||
```
|
|
||||||
|
|
||||||
Selecting a single module also runs its transitive dependencies. If a dependency
|
|
||||||
is unavailable or the dependency graph contains a cycle, the command logs a
|
|
||||||
warning and does not run any modules.
|
|
||||||
|
|
||||||
## Custom modules
|
|
||||||
|
|
||||||
Module-running `check-*` commands can load custom modules from Python files that
|
|
||||||
are not installed as part of MVT. Load one file with:
|
|
||||||
|
|
||||||
```bash
|
|
||||||
mvt-ios check-backup --load-module ./example_module.py --output ./out ./backup
|
|
||||||
```
|
|
||||||
|
|
||||||
You can also load a folder. MVT loads non-hidden top-level `*.py` files in
|
|
||||||
sorted order and skips `__init__.py`:
|
|
||||||
|
|
||||||
```bash
|
|
||||||
mvt-ios check-fs --load-module ./custom_modules ./filesystem-dump
|
|
||||||
```
|
|
||||||
|
|
||||||
Set `MVT_CUSTOM_MODULES` to load a folder for every module-running command. This
|
|
||||||
folder is loaded before any `--load-module` path:
|
|
||||||
|
|
||||||
```bash
|
|
||||||
MVT_CUSTOM_MODULES=./custom_modules mvt-android check-bugreport ./bugreport.zip
|
|
||||||
```
|
|
||||||
|
|
||||||
Custom modules are normal `MVTModule` subclasses:
|
|
||||||
|
|
||||||
```python
|
|
||||||
from mvt.common.module import MVTModule
|
|
||||||
|
|
||||||
|
|
||||||
class ExampleCustomModule(MVTModule):
|
|
||||||
supported_commands = (("ios", "check-backup"), ("ios", "check-fs"))
|
|
||||||
slug = "example_custom_module"
|
|
||||||
|
|
||||||
def run(self):
|
|
||||||
self.results = [{"message": "custom module ran"}]
|
|
||||||
|
|
||||||
def check_indicators(self):
|
|
||||||
pass
|
|
||||||
|
|
||||||
def serialize(self, result):
|
|
||||||
return None
|
|
||||||
```
|
|
||||||
|
|
||||||
Use `supported_commands` to 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:
|
|
||||||
|
|
||||||
```python
|
|
||||||
("ios", "check-backup")
|
|
||||||
("ios", "check-fs")
|
|
||||||
("ios", "check-iocs")
|
|
||||||
("ios", "check-sysdiagnose")
|
|
||||||
("android", "check-backup")
|
|
||||||
("android", "check-bugreport")
|
|
||||||
("android", "check-androidqf")
|
|
||||||
("android", "check-intrusion-logs")
|
|
||||||
("android", "check-iocs")
|
|
||||||
```
|
|
||||||
|
|
||||||
Custom modules can depend on existing MVT module classes. Dependencies are
|
|
||||||
resolved with the same ordering logic as built-in modules, and custom modules
|
|
||||||
are appended after built-ins before ordering:
|
|
||||||
|
|
||||||
```python
|
|
||||||
from mvt.common.module import MVTModule
|
|
||||||
from mvt.ios.modules.backup.manifest import Manifest
|
|
||||||
|
|
||||||
|
|
||||||
class DependentCustomModule(MVTModule):
|
|
||||||
supported_commands = (("ios", "check-backup"),)
|
|
||||||
dependencies = (Manifest,)
|
|
||||||
|
|
||||||
def run(self):
|
|
||||||
manifest_results = self.get_dependency_results(Manifest)
|
|
||||||
self.results = [{"manifest_entries": len(manifest_results)}]
|
|
||||||
```
|
|
||||||
|
|
||||||
## Profiling
|
|
||||||
|
|
||||||
Some MVT modules extract and process significant amounts of data during the analysis process or while checking results against known indicators. Care must be
|
|
||||||
take to avoid inefficient code paths as we add new modules.
|
|
||||||
|
|
||||||
MVT modules can be profiled with Python built-in `cProfile` by setting the `MVT_PROFILE` environment variable.
|
|
||||||
|
|
||||||
```bash
|
|
||||||
MVT_PROFILE=1 dev/mvt-ios check-backup test_backup
|
|
||||||
```
|
|
||||||
|
|
||||||
Open an issue or PR if you are encountering significant performance issues when analyzing a device with MVT.
|
|
||||||
@@ -1,7 +1,7 @@
|
|||||||
# Custom CLI Commands
|
# Custom CLI Commands
|
||||||
|
|
||||||
MVT can load additional top-level commands into `mvt-ios` and `mvt-android`.
|
MVT can load additional top-level commands into `mvt-ios` and `mvt-android`.
|
||||||
Custom commands are different from [custom forensic modules](development.md#custom-modules):
|
Custom commands are different from [custom forensic modules](index.md#custom-modules):
|
||||||
commands add new CLI operations, while modules add analysis steps to existing
|
commands add new CLI operations, while modules add analysis steps to existing
|
||||||
`check-*` commands.
|
`check-*` commands.
|
||||||
|
|
||||||
@@ -56,6 +56,10 @@ pipx inject mvt my-mvt-plugin
|
|||||||
When MVT is installed in an active virtual environment, install the plugin with
|
When MVT is installed in an active virtual environment, install the plugin with
|
||||||
`pip` in that environment.
|
`pip` in that environment.
|
||||||
|
|
||||||
|
Command packages that need their own settings, such as an API key, should store
|
||||||
|
them in a namespaced [plugin configuration file](plugin_configuration.md)
|
||||||
|
rather than in MVT's own `config.yaml`.
|
||||||
|
|
||||||
## Load a Command File
|
## Load a Command File
|
||||||
|
|
||||||
For local commands that are not packaged, create a Python file that exports one
|
For local commands that are not packaged, create a Python file that exports one
|
||||||
@@ -0,0 +1,229 @@
|
|||||||
|
# Development
|
||||||
|
|
||||||
|
The Mobile Verification Toolkit team welcomes contributions of new forensic modules or other contributions which help improve the software.
|
||||||
|
|
||||||
|
## Local environment
|
||||||
|
|
||||||
|
MVT uses `uv` for dependency management. To install the project and development dependencies from the locked environment, run:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
make install
|
||||||
|
```
|
||||||
|
|
||||||
|
## Testing
|
||||||
|
|
||||||
|
MVT uses `pytest` for unit and integration tests. Code style consistency is maintained with `ruff` and `mypy`. All can
|
||||||
|
be run automatically with:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
make check
|
||||||
|
```
|
||||||
|
|
||||||
|
Run these tests before making new commits or opening pull requests.
|
||||||
|
|
||||||
|
## Module dependencies
|
||||||
|
|
||||||
|
Modules can require other modules to run first by declaring their classes in
|
||||||
|
`dependencies`. The command runner uses a stable topological ordering, so the
|
||||||
|
existing module list order is preserved wherever dependency constraints allow.
|
||||||
|
|
||||||
|
```python
|
||||||
|
class DependentModule(MVTModule):
|
||||||
|
dependencies = (PrerequisiteModule,)
|
||||||
|
|
||||||
|
def run(self):
|
||||||
|
prerequisite_results = self.get_dependency_results(PrerequisiteModule)
|
||||||
|
```
|
||||||
|
|
||||||
|
Selecting a single module also runs its transitive dependencies.
|
||||||
|
|
||||||
|
A module can only depend on modules the command it runs in also has. When a
|
||||||
|
declared dependency is not among them, the command logs a warning naming the
|
||||||
|
module and the missing dependency, skips that module and everything depending
|
||||||
|
on it, and runs the rest of the analysis. Selecting such a module with
|
||||||
|
`--module` therefore leaves nothing to run, which the warning explains.
|
||||||
|
|
||||||
|
A cycle in the dependency graph is a programming error rather than a
|
||||||
|
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:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
mvt-ios check-backup --load-module ./example_module.py --output ./out ./backup
|
||||||
|
```
|
||||||
|
|
||||||
|
You can also load a folder. MVT loads non-hidden top-level `*.py` files in
|
||||||
|
sorted order and skips `__init__.py`:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
mvt-ios check-fs --load-module ./custom_modules ./filesystem-dump
|
||||||
|
```
|
||||||
|
|
||||||
|
Set `MVT_CUSTOM_MODULES` to load a folder for every module-running command. This
|
||||||
|
folder is loaded before any `--load-module` path:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
MVT_CUSTOM_MODULES=./custom_modules mvt-android check-bugreport ./bugreport.zip
|
||||||
|
```
|
||||||
|
|
||||||
|
Custom modules are normal `MVTModule` subclasses:
|
||||||
|
|
||||||
|
```python
|
||||||
|
from mvt.common.module import MVTModule
|
||||||
|
|
||||||
|
|
||||||
|
class ExampleCustomModule(MVTModule):
|
||||||
|
supported_commands = (("ios", "check-backup"), ("ios", "check-fs"))
|
||||||
|
slug = "example_custom_module"
|
||||||
|
|
||||||
|
def run(self):
|
||||||
|
self.results = [{"message": "custom module ran"}]
|
||||||
|
|
||||||
|
def check_indicators(self):
|
||||||
|
pass
|
||||||
|
|
||||||
|
def serialize(self, result):
|
||||||
|
return None
|
||||||
|
```
|
||||||
|
|
||||||
|
Use `supported_commands` to 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:
|
||||||
|
|
||||||
|
```python
|
||||||
|
("ios", "check-backup")
|
||||||
|
("ios", "check-fs")
|
||||||
|
("ios", "check-iocs")
|
||||||
|
("ios", "check-sysdiagnose")
|
||||||
|
("android", "check-backup")
|
||||||
|
("android", "check-bugreport")
|
||||||
|
("android", "check-androidqf")
|
||||||
|
("android", "check-intrusion-logs")
|
||||||
|
("android", "check-iocs")
|
||||||
|
```
|
||||||
|
|
||||||
|
Custom modules can depend on existing MVT module classes. Dependencies are
|
||||||
|
resolved with the same ordering logic as built-in modules, and custom modules
|
||||||
|
are appended after built-ins before ordering:
|
||||||
|
|
||||||
|
```python
|
||||||
|
from mvt.common.module import MVTModule
|
||||||
|
from mvt.ios.modules.backup.manifest import Manifest
|
||||||
|
|
||||||
|
|
||||||
|
class DependentCustomModule(MVTModule):
|
||||||
|
supported_commands = (("ios", "check-backup"),)
|
||||||
|
dependencies = (Manifest,)
|
||||||
|
|
||||||
|
def run(self):
|
||||||
|
manifest_results = self.get_dependency_results(Manifest)
|
||||||
|
self.results = [{"manifest_entries": len(manifest_results)}]
|
||||||
|
```
|
||||||
|
|
||||||
|
## Installed module packages
|
||||||
|
|
||||||
|
Python packages can register modules so they load automatically in every
|
||||||
|
module-running `check-*` command, without `--load-module` or
|
||||||
|
`MVT_CUSTOM_MODULES`. Register an entry point in the `mvt.modules` group in
|
||||||
|
the package's `pyproject.toml`:
|
||||||
|
|
||||||
|
```toml
|
||||||
|
[project.entry-points."mvt.modules"]
|
||||||
|
mvt-plugin-amnesty-custom = "mvt_plugin_amnesty_custom: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
|
||||||
|
|
||||||
|
|
||||||
|
class PackagedModule(MVTModule):
|
||||||
|
supported_commands = (("ios", "check-backup"),)
|
||||||
|
|
||||||
|
def run(self):
|
||||||
|
self.results = [{"message": "packaged module ran"}]
|
||||||
|
|
||||||
|
|
||||||
|
def get_modules() -> list[type[MVTModule]]:
|
||||||
|
return [PackagedModule]
|
||||||
|
```
|
||||||
|
|
||||||
|
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
|
||||||
|
does not prevent MVT from running. As with custom commands, installed module
|
||||||
|
packages run as trusted code inside the MVT process, so install only packages
|
||||||
|
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
|
||||||
|
```
|
||||||
|
|
||||||
|
Module packages that need their own settings, such as an API key, should store
|
||||||
|
them in a namespaced [plugin configuration file](plugin_configuration.md)
|
||||||
|
rather than in MVT's own `config.yaml`.
|
||||||
|
|
||||||
|
### Naming module packages
|
||||||
|
|
||||||
|
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`.
|
||||||
|
|
||||||
|
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
|
||||||
|
technical requirement: modules load through the `mvt.modules` entry point
|
||||||
|
regardless of what the package is called, and existing packages with other
|
||||||
|
names keep working. Note that the prefix is also not a mark of authenticity —
|
||||||
|
anyone can publish a package with any available name, so vet a module package
|
||||||
|
and its publisher before installing it, whatever it is called.
|
||||||
|
|
||||||
|
### Module logging
|
||||||
|
|
||||||
|
Modules log through `self.log`, and MVT names the logger for where the module
|
||||||
|
came from. MVT's own modules log under their dotted path (for example
|
||||||
|
`mvt.ios.modules.mixed.whatsapp`). Everything external is namespaced under
|
||||||
|
`mvt.ext` to keep it visually distinct from built-in modules and isolated from
|
||||||
|
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.*`.
|
||||||
|
- Files loaded with `--load-module` or `MVT_CUSTOM_MODULES` log as
|
||||||
|
`mvt.ext.<file name>`.
|
||||||
|
|
||||||
|
## Auditing loaded modules
|
||||||
|
|
||||||
|
Because installed module packages load automatically, MVT records where every
|
||||||
|
module came from:
|
||||||
|
|
||||||
|
- `--list-modules` groups the available modules by source: MVT itself
|
||||||
|
(with its version), each installed package (with its version and, when
|
||||||
|
installed directly from a repository, the commit), and each file loaded
|
||||||
|
with `--load-module` or `MVT_CUSTOM_MODULES` (with the SHA-256 hash of the
|
||||||
|
file).
|
||||||
|
- When a command runs with an `--output` folder, the `command.log` file
|
||||||
|
records one line per module source with the source's version or hash and
|
||||||
|
the list of modules loaded from it.
|
||||||
|
|
||||||
|
## Profiling
|
||||||
|
|
||||||
|
Some MVT modules extract and process significant amounts of data during the analysis process or while checking results against known indicators. Care must be
|
||||||
|
take to avoid inefficient code paths as we add new modules.
|
||||||
|
|
||||||
|
MVT modules can be profiled with Python built-in `cProfile` by setting the `MVT_PROFILE` environment variable.
|
||||||
|
|
||||||
|
```bash
|
||||||
|
MVT_PROFILE=1 dev/mvt-ios check-backup test_backup
|
||||||
|
```
|
||||||
|
|
||||||
|
Open an issue or PR if you are encountering significant performance issues when analyzing a device with MVT.
|
||||||
@@ -0,0 +1,149 @@
|
|||||||
|
# Plugin Configuration
|
||||||
|
|
||||||
|
Plugin packages that add [custom CLI commands](custom_commands.md) or
|
||||||
|
[modules](index.md#installed-module-packages) often need to store their
|
||||||
|
own settings, such as an API key, a server URL or the timestamp of the last
|
||||||
|
synchronization. MVT provides a namespaced settings base class so each plugin
|
||||||
|
keeps its configuration in its own file, and a data folder for anything else a
|
||||||
|
plugin needs to keep on disk.
|
||||||
|
|
||||||
|
!!! warning
|
||||||
|
|
||||||
|
Do not write plugin settings to MVT's own `config.yaml`. MVT rewrites that
|
||||||
|
file with the settings it knows about every time it starts, so any other
|
||||||
|
section is deleted.
|
||||||
|
|
||||||
|
## Where Settings Are Stored
|
||||||
|
|
||||||
|
Each plugin gets one YAML file in a `plugins` folder next to MVT's own
|
||||||
|
configuration:
|
||||||
|
|
||||||
|
```
|
||||||
|
~/.config/mvt/plugins/<plugin name>.yaml
|
||||||
|
```
|
||||||
|
|
||||||
|
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 names must be lowercase and may only contain letters, digits and dashes,
|
||||||
|
matching the `mvt-plugin-<name>` package naming convention. MVT creates the
|
||||||
|
`plugins` folder with `0700` permissions and writes the settings files with
|
||||||
|
`0600` permissions, because they commonly hold credentials. Files are written
|
||||||
|
through a temporary file and moved into place, so an interrupted save never
|
||||||
|
leaves a partially written settings file behind.
|
||||||
|
|
||||||
|
## Plugin Data Folder
|
||||||
|
|
||||||
|
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:
|
||||||
|
|
||||||
|
```
|
||||||
|
~/.local/share/mvt/plugin-data/<plugin name>/ # Linux
|
||||||
|
~/Library/Application Support/mvt/plugin-data/<plugin name>/ # macOS
|
||||||
|
```
|
||||||
|
|
||||||
|
The folder sits beside MVT's own data, such as the downloaded indicators. It is
|
||||||
|
created if it is missing, with `0700` permissions. Asking for it again returns
|
||||||
|
the same path and leaves the contents alone, so a plugin can ask for it every
|
||||||
|
time it needs the folder. `ExamplePluginSettings` below is the settings class
|
||||||
|
defined in the next section:
|
||||||
|
|
||||||
|
```python
|
||||||
|
import os
|
||||||
|
|
||||||
|
|
||||||
|
def cache_path() -> str:
|
||||||
|
folder = ExamplePluginSettings.data_folder()
|
||||||
|
return os.path.join(folder, "virustotal_lookups_cache.json")
|
||||||
|
```
|
||||||
|
|
||||||
|
A plugin which has no settings class calls
|
||||||
|
`plugin_data_folder("example-plugin")` instead.
|
||||||
|
|
||||||
|
Do not fall back on a path of your own such as `~/.cache/example-plugin`: it
|
||||||
|
is a Linux-only convention, and MVT will not create it for you.
|
||||||
|
|
||||||
|
## Defining Plugin Settings
|
||||||
|
|
||||||
|
Subclass `MVTPluginSettings`, set `plugin_name` and declare typed fields with
|
||||||
|
defaults:
|
||||||
|
|
||||||
|
```python
|
||||||
|
from typing import Optional
|
||||||
|
|
||||||
|
from mvt.common.plugin_config import MVTPluginSettings
|
||||||
|
|
||||||
|
|
||||||
|
class ExamplePluginSettings(MVTPluginSettings):
|
||||||
|
plugin_name = "example-plugin"
|
||||||
|
|
||||||
|
API_KEY: Optional[str] = None
|
||||||
|
MAX_RESULTS: int = 25
|
||||||
|
LAST_SYNC: Optional[str] = None
|
||||||
|
```
|
||||||
|
|
||||||
|
`load()` returns the current settings and `save()` writes them back:
|
||||||
|
|
||||||
|
```python
|
||||||
|
from datetime import datetime, timezone
|
||||||
|
|
||||||
|
import click
|
||||||
|
|
||||||
|
|
||||||
|
def sync():
|
||||||
|
settings = ExamplePluginSettings.load()
|
||||||
|
if not settings.API_KEY:
|
||||||
|
raise click.ClickException(
|
||||||
|
"No API key configured. Set MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY or "
|
||||||
|
"run 'example-plugin configure'."
|
||||||
|
)
|
||||||
|
|
||||||
|
settings.LAST_SYNC = datetime.now(timezone.utc).isoformat()
|
||||||
|
settings.save()
|
||||||
|
```
|
||||||
|
|
||||||
|
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`.
|
||||||
|
A settings file that cannot be parsed, or that does not hold a mapping of
|
||||||
|
setting names to values, raises a `PluginConfigLoadError` naming the file.
|
||||||
|
|
||||||
|
Every subclass that sets its own `plugin_name` gets its own file and its own
|
||||||
|
environment namespace. A subclass that does not redefine `plugin_name` inherits
|
||||||
|
it, and therefore shares the file and the environment variables of its parent
|
||||||
|
class.
|
||||||
|
|
||||||
|
## Environment Variables
|
||||||
|
|
||||||
|
Every field can also be set with an environment variable. The prefix is
|
||||||
|
`MVT_PLUGIN_`, followed by the plugin name upper-cased with dashes replaced by
|
||||||
|
underscores, followed by the field name. For the example above:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
export MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY=...
|
||||||
|
export MVT_PLUGIN_EXAMPLE_PLUGIN_MAX_RESULTS=50
|
||||||
|
```
|
||||||
|
|
||||||
|
Settings resolve in this order, from highest to lowest priority:
|
||||||
|
|
||||||
|
1. Arguments passed to the settings class directly, such as
|
||||||
|
`ExamplePluginSettings(API_KEY="...")`
|
||||||
|
2. Environment variables
|
||||||
|
3. The plugin's YAML file
|
||||||
|
4. The field defaults declared on the settings class
|
||||||
|
|
||||||
|
!!! tip
|
||||||
|
|
||||||
|
On shared or multi-user machines, prefer passing API keys through
|
||||||
|
environment variables rather than saving them to the plugin file. `save()`
|
||||||
|
skips every value that the environment currently supplies, so a credential
|
||||||
|
provided that way is not copied into the settings file when a plugin saves
|
||||||
|
an unrelated setting.
|
||||||
|
|
||||||
|
Unknown keys in a plugin's YAML file are ignored, so a settings file written by
|
||||||
|
a newer version of a plugin does not break an older one.
|
||||||
+3
-2
@@ -50,6 +50,7 @@ nav:
|
|||||||
- Check Android Intrusion Logs: "android/intrusion_logs.md"
|
- Check Android Intrusion Logs: "android/intrusion_logs.md"
|
||||||
- Indicators of Compromise: "iocs.md"
|
- Indicators of Compromise: "iocs.md"
|
||||||
- Development:
|
- Development:
|
||||||
- Development Instructions: "development.md"
|
- Development Instructions: "development/index.md"
|
||||||
- Custom CLI Commands: "custom_commands.md"
|
- Custom CLI Commands: "development/custom_commands.md"
|
||||||
|
- Plugin Configuration: "development/plugin_configuration.md"
|
||||||
- License: "license.md"
|
- License: "license.md"
|
||||||
@@ -9,6 +9,7 @@ from typing import Optional
|
|||||||
|
|
||||||
from mvt.common.command import Command
|
from mvt.common.command import Command
|
||||||
from mvt.common.module import MVTModule
|
from mvt.common.module import MVTModule
|
||||||
|
from mvt.common.module_loader import get_module_logger
|
||||||
from mvt.common.utils import exec_or_profile
|
from mvt.common.utils import exec_or_profile
|
||||||
|
|
||||||
log = logging.getLogger(__name__)
|
log = logging.getLogger(__name__)
|
||||||
@@ -76,7 +77,7 @@ class CmdCheckIOCS(Command):
|
|||||||
)
|
)
|
||||||
|
|
||||||
m = iocs_module.from_json(
|
m = iocs_module.from_json(
|
||||||
file_path, log=logging.getLogger(iocs_module.__module__)
|
file_path, log=get_module_logger(iocs_module)
|
||||||
)
|
)
|
||||||
if not m:
|
if not m:
|
||||||
log.warning("No result from this module, skipping it")
|
log.warning("No result from this module, skipping it")
|
||||||
|
|||||||
+130
-20
@@ -19,7 +19,12 @@ from .alerts import AlertLevel, AlertStore
|
|||||||
from .config import settings
|
from .config import settings
|
||||||
from .indicators import Indicators
|
from .indicators import Indicators
|
||||||
from .module import EncryptedBackupError, MVTModule, run_module, save_timeline
|
from .module import EncryptedBackupError, MVTModule, run_module, save_timeline
|
||||||
from .module_loader import module_supports_command
|
from .module_loader import (
|
||||||
|
ModuleOrigin,
|
||||||
|
get_module_logger,
|
||||||
|
get_module_origin,
|
||||||
|
module_supports_command,
|
||||||
|
)
|
||||||
from .module_types import ModuleTimeline, URLResult
|
from .module_types import ModuleTimeline, URLResult
|
||||||
from .utils import (
|
from .utils import (
|
||||||
CustomJSONEncoder,
|
CustomJSONEncoder,
|
||||||
@@ -210,10 +215,34 @@ class Command:
|
|||||||
for file in generate_hashes_from_path(self.target_path, self.log):
|
for file in generate_hashes_from_path(self.target_path, self.log):
|
||||||
self.hash_values.append(file)
|
self.hash_values.append(file)
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _modules_by_origin(
|
||||||
|
modules: list[type[MVTModule]],
|
||||||
|
) -> dict[ModuleOrigin, list[str]]:
|
||||||
|
origins: dict[ModuleOrigin, list[str]] = {}
|
||||||
|
for module in modules:
|
||||||
|
origins.setdefault(get_module_origin(module), []).append(module.__name__)
|
||||||
|
return origins
|
||||||
|
|
||||||
def list_modules(self) -> None:
|
def list_modules(self) -> None:
|
||||||
self.log.info("Following is the list of available %s modules:", self.name)
|
self.log.info("Following is the list of available %s modules:", self.name)
|
||||||
for module in self._available_modules():
|
for origin, module_names in self._modules_by_origin(
|
||||||
self.log.info(" - %s", module.__name__)
|
self._available_modules()
|
||||||
|
).items():
|
||||||
|
self.log.info(
|
||||||
|
" - Modules from %s: %s", origin.label, ", ".join(module_names)
|
||||||
|
)
|
||||||
|
|
||||||
|
def _log_loaded_modules(self, modules: list[type[MVTModule]]) -> None:
|
||||||
|
"""Record the loaded modules and their origins for auditability."""
|
||||||
|
for origin, module_names in self._modules_by_origin(modules).items():
|
||||||
|
self.log.info(
|
||||||
|
"Loaded %d %s modules from %s: %s",
|
||||||
|
len(module_names),
|
||||||
|
self.name,
|
||||||
|
origin.label,
|
||||||
|
", ".join(module_names),
|
||||||
|
)
|
||||||
|
|
||||||
def _available_modules(self) -> list[type[MVTModule]]:
|
def _available_modules(self) -> list[type[MVTModule]]:
|
||||||
modules = list(self.modules)
|
modules = list(self.modules)
|
||||||
@@ -288,6 +317,81 @@ class Command:
|
|||||||
console.print("")
|
console.print("")
|
||||||
console.print(panel)
|
console.print(panel)
|
||||||
|
|
||||||
|
def _skipped_modules(
|
||||||
|
self,
|
||||||
|
required: list[type[MVTModule]],
|
||||||
|
module_indexes: dict[type[MVTModule], int],
|
||||||
|
) -> dict[type[MVTModule], tuple[type[MVTModule], type[MVTModule]]]:
|
||||||
|
"""Return the modules to drop because a dependency is unavailable.
|
||||||
|
|
||||||
|
A module declaring a dependency this command cannot provide is unable
|
||||||
|
to run, and so is every module depending on it. Dropping only those
|
||||||
|
keeps a single wrong declaration - in a module scoped to several
|
||||||
|
commands, for example - from silencing an entire analysis.
|
||||||
|
|
||||||
|
The returned mapping gives, for each skipped module, the module which
|
||||||
|
is missing a dependency and the dependency it is missing.
|
||||||
|
"""
|
||||||
|
skipped: dict[type[MVTModule], tuple[type[MVTModule], type[MVTModule]]] = {}
|
||||||
|
# Skipping the one module a run was asked for leaves nothing to run,
|
||||||
|
# which the caller reports instead.
|
||||||
|
remainder = (
|
||||||
|
"" if self.module_name else " The rest of the analysis will still run."
|
||||||
|
)
|
||||||
|
|
||||||
|
# Skipping one module can skip the modules depending on it, which the
|
||||||
|
# pass over the module list may already have gone past, so repeat the
|
||||||
|
# pass until nothing changes.
|
||||||
|
changed = True
|
||||||
|
while changed:
|
||||||
|
changed = False
|
||||||
|
for module in required:
|
||||||
|
if module in skipped:
|
||||||
|
continue
|
||||||
|
|
||||||
|
for dependency in module.dependencies:
|
||||||
|
if dependency not in module_indexes:
|
||||||
|
skipped[module] = (module, dependency)
|
||||||
|
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__,
|
||||||
|
remainder,
|
||||||
|
)
|
||||||
|
break
|
||||||
|
|
||||||
|
if dependency in skipped:
|
||||||
|
root, missing = skipped[dependency]
|
||||||
|
skipped[module] = (root, missing)
|
||||||
|
changed = True
|
||||||
|
if dependency is root:
|
||||||
|
self.log.warning(
|
||||||
|
"Module %s will be SKIPPED: it depends on "
|
||||||
|
"module %s, itself skipped for depending on "
|
||||||
|
"unavailable module %s.%s",
|
||||||
|
module.__name__,
|
||||||
|
dependency.__name__,
|
||||||
|
missing.__name__,
|
||||||
|
remainder,
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
self.log.warning(
|
||||||
|
"Module %s will be SKIPPED: it depends on "
|
||||||
|
"skipped module %s, in a chain starting at "
|
||||||
|
"module %s, which depends on unavailable "
|
||||||
|
"module %s.%s",
|
||||||
|
module.__name__,
|
||||||
|
dependency.__name__,
|
||||||
|
root.__name__,
|
||||||
|
missing.__name__,
|
||||||
|
remainder,
|
||||||
|
)
|
||||||
|
break
|
||||||
|
|
||||||
|
return skipped
|
||||||
|
|
||||||
def _ordered_modules(self) -> Optional[list[type[MVTModule]]]:
|
def _ordered_modules(self) -> Optional[list[type[MVTModule]]]:
|
||||||
"""Return enabled modules in stable topological order."""
|
"""Return enabled modules in stable topological order."""
|
||||||
modules = self._available_modules()
|
modules = self._available_modules()
|
||||||
@@ -300,30 +404,34 @@ class Command:
|
|||||||
else:
|
else:
|
||||||
selected = [module for module in modules if module.enabled]
|
selected = [module for module in modules if module.enabled]
|
||||||
|
|
||||||
required = set(selected)
|
required: set[type[MVTModule]] = set()
|
||||||
pending = list(selected)
|
pending = list(selected)
|
||||||
while pending:
|
while pending:
|
||||||
module = pending.pop()
|
module = pending.pop()
|
||||||
|
if module in required:
|
||||||
|
continue
|
||||||
|
required.add(module)
|
||||||
for dependency in module.dependencies:
|
for dependency in module.dependencies:
|
||||||
if dependency not in module_indexes:
|
# Unavailable dependencies are reported by _skipped_modules().
|
||||||
self.log.warning(
|
if dependency in module_indexes:
|
||||||
"Module %s depends on unavailable module %s. "
|
|
||||||
"No modules will be run.",
|
|
||||||
module.__name__,
|
|
||||||
dependency.__name__,
|
|
||||||
)
|
|
||||||
return None
|
|
||||||
if dependency not in required:
|
|
||||||
required.add(dependency)
|
|
||||||
pending.append(dependency)
|
pending.append(dependency)
|
||||||
|
|
||||||
|
ordered_required = sorted(required, key=lambda module: module_indexes[module])
|
||||||
|
skipped = self._skipped_modules(ordered_required, module_indexes)
|
||||||
|
runnable = [module for module in ordered_required if module not in skipped]
|
||||||
|
if skipped and not runnable:
|
||||||
|
self.log.warning(
|
||||||
|
"Every selected module was skipped for an unavailable "
|
||||||
|
"dependency. No modules will be run."
|
||||||
|
)
|
||||||
|
|
||||||
dependents: dict[type[MVTModule], list[type[MVTModule]]] = {
|
dependents: dict[type[MVTModule], list[type[MVTModule]]] = {
|
||||||
module: [] for module in required
|
module: [] for module in runnable
|
||||||
}
|
}
|
||||||
indegree = {module: 0 for module in required}
|
indegree = {module: 0 for module in runnable}
|
||||||
for module in required:
|
for module in runnable:
|
||||||
for dependency in module.dependencies:
|
for dependency in module.dependencies:
|
||||||
if dependency not in required:
|
if dependency not in indegree:
|
||||||
continue
|
continue
|
||||||
dependents[dependency].append(module)
|
dependents[dependency].append(module)
|
||||||
indegree[module] += 1
|
indegree[module] += 1
|
||||||
@@ -342,7 +450,7 @@ class Command:
|
|||||||
if indegree[dependent] == 0:
|
if indegree[dependent] == 0:
|
||||||
heappush(ready, (module_indexes[dependent], dependent))
|
heappush(ready, (module_indexes[dependent], dependent))
|
||||||
|
|
||||||
if len(ordered) != len(required):
|
if len(ordered) != len(runnable):
|
||||||
cyclic_modules = sorted(
|
cyclic_modules = sorted(
|
||||||
(module.__name__ for module, count in indegree.items() if count > 0)
|
(module.__name__ for module, count in indegree.items() if count > 0)
|
||||||
)
|
)
|
||||||
@@ -360,6 +468,8 @@ class Command:
|
|||||||
if ordered_modules is None:
|
if ordered_modules is None:
|
||||||
return
|
return
|
||||||
|
|
||||||
|
self._log_loaded_modules(ordered_modules)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
self.init()
|
self.init()
|
||||||
except NotImplementedError:
|
except NotImplementedError:
|
||||||
@@ -368,7 +478,7 @@ class Command:
|
|||||||
executed_by_type: dict[type[MVTModule], MVTModule] = {}
|
executed_by_type: dict[type[MVTModule], MVTModule] = {}
|
||||||
for module in ordered_modules:
|
for module in ordered_modules:
|
||||||
|
|
||||||
module_logger = logging.getLogger(module.__module__)
|
module_logger = get_module_logger(module)
|
||||||
|
|
||||||
m = module(
|
m = module(
|
||||||
target_path=self.target_path,
|
target_path=self.target_path,
|
||||||
|
|||||||
@@ -8,7 +8,7 @@ HELP_MSG_VERSION = "Show the currently installed version of MVT"
|
|||||||
HELP_MSG_OUTPUT = "Specify a path to a folder where you want to store JSON results"
|
HELP_MSG_OUTPUT = "Specify a path to a folder where you want to store JSON results"
|
||||||
HELP_MSG_IOC = "Path to indicators file (can be invoked multiple time)"
|
HELP_MSG_IOC = "Path to indicators file (can be invoked multiple time)"
|
||||||
HELP_MSG_FAST = "Avoid running time/resource consuming features"
|
HELP_MSG_FAST = "Avoid running time/resource consuming features"
|
||||||
HELP_MSG_LIST_MODULES = "Print list of available modules and exit"
|
HELP_MSG_LIST_MODULES = "Print list of available modules and their source, then exit"
|
||||||
HELP_MSG_MODULE = "Name of a single module you would like to run instead of all"
|
HELP_MSG_MODULE = "Name of a single module you would like to run instead of all"
|
||||||
HELP_MSG_LOAD_MODULE = (
|
HELP_MSG_LOAD_MODULE = (
|
||||||
"Load custom MVT module(s) from a Python file or folder "
|
"Load custom MVT module(s) from a Python file or folder "
|
||||||
|
|||||||
@@ -4,18 +4,28 @@
|
|||||||
# https://license.mvt.re/1.1/
|
# https://license.mvt.re/1.1/
|
||||||
|
|
||||||
import hashlib
|
import hashlib
|
||||||
|
import importlib.metadata
|
||||||
import importlib.util
|
import importlib.util
|
||||||
import inspect
|
import inspect
|
||||||
|
import json
|
||||||
import logging
|
import logging
|
||||||
import os
|
import os
|
||||||
import sys
|
import sys
|
||||||
|
from dataclasses import dataclass
|
||||||
|
from functools import lru_cache
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from types import ModuleType
|
from types import ModuleType
|
||||||
from typing import Iterable, Optional
|
from typing import Iterable, Optional
|
||||||
|
|
||||||
from .module import MVTModule
|
from .module import MVTModule
|
||||||
|
from .version import MVT_VERSION
|
||||||
|
|
||||||
MVT_CUSTOM_MODULES_ENV = "MVT_CUSTOM_MODULES"
|
MVT_CUSTOM_MODULES_ENV = "MVT_CUSTOM_MODULES"
|
||||||
|
MODULES_ENTRY_POINT_GROUP = "mvt.modules"
|
||||||
|
EXTERNAL_LOGGER_NAMESPACE = "mvt.ext"
|
||||||
|
PLUGIN_PACKAGE_PREFIX = "mvt_plugin_"
|
||||||
|
_ORIGIN_ATTRIBUTE = "_mvt_module_origin"
|
||||||
|
_PATH_MODULE_PREFIX = "_mvt_custom_module_"
|
||||||
log = logging.getLogger(__name__)
|
log = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
@@ -23,15 +33,74 @@ class CustomModuleLoadError(Exception):
|
|||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class ModuleOrigin:
|
||||||
|
"""Describes where a loaded module came from, for auditability.
|
||||||
|
|
||||||
|
``kind`` is one of ``builtin`` (shipped with MVT), ``package`` (loaded
|
||||||
|
from an installed package) or ``path`` (loaded from a file passed with
|
||||||
|
``--load-module`` or the environment variable).
|
||||||
|
"""
|
||||||
|
|
||||||
|
kind: str
|
||||||
|
name: str
|
||||||
|
version: Optional[str] = None
|
||||||
|
commit: Optional[str] = None
|
||||||
|
file_sha256: Optional[str] = None
|
||||||
|
|
||||||
|
@property
|
||||||
|
def label(self) -> str:
|
||||||
|
label = self.name
|
||||||
|
if self.version:
|
||||||
|
label += f"@{self.version}"
|
||||||
|
label = f"'{label}'"
|
||||||
|
if self.commit:
|
||||||
|
label += f" (commit {self.commit})"
|
||||||
|
if self.file_sha256:
|
||||||
|
label += f" (sha256: {self.file_sha256})"
|
||||||
|
return label
|
||||||
|
|
||||||
|
|
||||||
def _module_name_for_path(path: Path) -> str:
|
def _module_name_for_path(path: Path) -> str:
|
||||||
digest = hashlib.sha256(str(path).encode("utf-8")).hexdigest()[:16]
|
digest = hashlib.sha256(str(path).encode("utf-8")).hexdigest()[:16]
|
||||||
return f"_mvt_custom_module_{path.stem}_{digest}"
|
return f"{_PATH_MODULE_PREFIX}{path.stem}_{digest}"
|
||||||
|
|
||||||
|
|
||||||
|
def get_module_logger(module_class: type[MVTModule]) -> logging.Logger:
|
||||||
|
"""Return the logger a module's records should be emitted through.
|
||||||
|
|
||||||
|
Modules loaded from installed packages or file paths live outside the
|
||||||
|
"mvt" logger hierarchy, so their records would never reach the handlers
|
||||||
|
attached to the "mvt" logger and instead fall through to
|
||||||
|
logging.lastResort (which prints bare messages and drops anything below
|
||||||
|
WARNING). Their loggers are parented under the "mvt.ext" namespace,
|
||||||
|
keeping external module names from colliding with MVT's own logger
|
||||||
|
tree. File-path modules are named after their file instead of the
|
||||||
|
mangled internal import name, and packages following the recommended
|
||||||
|
"mvt_plugin_<name>" naming convention log under "mvt.ext.<name>".
|
||||||
|
"""
|
||||||
|
name = module_class.__module__
|
||||||
|
if name == "mvt" or name.startswith("mvt."):
|
||||||
|
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
|
||||||
|
|
||||||
|
return logging.getLogger(f"{EXTERNAL_LOGGER_NAMESPACE}.{name}")
|
||||||
|
|
||||||
|
|
||||||
def _iter_module_files(path: Path) -> Iterable[Path]:
|
def _iter_module_files(path: Path) -> Iterable[Path]:
|
||||||
if path.is_file():
|
if path.is_file():
|
||||||
if path.suffix != ".py":
|
if path.suffix != ".py":
|
||||||
raise CustomModuleLoadError(f"Custom module file is not a Python file: {path}")
|
raise CustomModuleLoadError(
|
||||||
|
f"Custom module file is not a Python file: {path}"
|
||||||
|
)
|
||||||
yield path
|
yield path
|
||||||
return
|
return
|
||||||
|
|
||||||
@@ -59,7 +128,9 @@ def _load_python_file(path: Path) -> ModuleType:
|
|||||||
try:
|
try:
|
||||||
spec.loader.exec_module(module)
|
spec.loader.exec_module(module)
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
raise CustomModuleLoadError(f"Unable to import custom module {path}: {exc}") from exc
|
raise CustomModuleLoadError(
|
||||||
|
f"Unable to import custom module {path}: {exc}"
|
||||||
|
) from exc
|
||||||
|
|
||||||
return module
|
return module
|
||||||
|
|
||||||
@@ -84,17 +155,156 @@ def load_custom_modules_from_path(path: str) -> list[type[MVTModule]]:
|
|||||||
resolved_path = Path(path).expanduser().resolve()
|
resolved_path = Path(path).expanduser().resolve()
|
||||||
|
|
||||||
for module_file in _iter_module_files(resolved_path):
|
for module_file in _iter_module_files(resolved_path):
|
||||||
|
file_sha256 = hashlib.sha256(module_file.read_bytes()).hexdigest()
|
||||||
loaded_module = _load_python_file(module_file)
|
loaded_module = _load_python_file(module_file)
|
||||||
|
origin = ModuleOrigin(
|
||||||
|
kind="path", name=str(module_file), file_sha256=file_sha256
|
||||||
|
)
|
||||||
for module_class in discover_mvt_modules(loaded_module):
|
for module_class in discover_mvt_modules(loaded_module):
|
||||||
key = (str(module_file), module_class.__qualname__)
|
key = (str(module_file), module_class.__qualname__)
|
||||||
if key in seen:
|
if key in seen:
|
||||||
continue
|
continue
|
||||||
seen.add(key)
|
seen.add(key)
|
||||||
|
setattr(module_class, _ORIGIN_ATTRIBUTE, origin)
|
||||||
custom_modules.append(module_class)
|
custom_modules.append(module_class)
|
||||||
|
|
||||||
return custom_modules
|
return custom_modules
|
||||||
|
|
||||||
|
|
||||||
|
def _module_key(module_class: type[MVTModule]) -> tuple[str, str]:
|
||||||
|
try:
|
||||||
|
source = str(Path(inspect.getfile(module_class)).resolve())
|
||||||
|
except (OSError, TypeError):
|
||||||
|
source = module_class.__module__
|
||||||
|
return (source, module_class.__qualname__)
|
||||||
|
|
||||||
|
|
||||||
|
def _distribution_commit(dist: importlib.metadata.Distribution) -> Optional[str]:
|
||||||
|
"""Return the VCS commit a distribution was installed from, if recorded.
|
||||||
|
|
||||||
|
Packages installed directly from a repository (``pip install git+...``)
|
||||||
|
record the commit in ``direct_url.json`` (PEP 610).
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
direct_url_text = dist.read_text("direct_url.json")
|
||||||
|
if not direct_url_text:
|
||||||
|
return None
|
||||||
|
commit = json.loads(direct_url_text).get("vcs_info", {}).get("commit_id")
|
||||||
|
return commit if isinstance(commit, str) else None
|
||||||
|
except Exception:
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def _entry_point_origin(entry_point: importlib.metadata.EntryPoint) -> ModuleOrigin:
|
||||||
|
name = entry_point.name
|
||||||
|
version = None
|
||||||
|
commit = None
|
||||||
|
# Manually constructed entry points have no associated distribution.
|
||||||
|
dist = getattr(entry_point, "dist", None)
|
||||||
|
if dist is not None:
|
||||||
|
try:
|
||||||
|
name = dist.name or name
|
||||||
|
version = dist.version
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
commit = _distribution_commit(dist)
|
||||||
|
return ModuleOrigin(kind="package", name=name, version=version, commit=commit)
|
||||||
|
|
||||||
|
|
||||||
|
@lru_cache(maxsize=1)
|
||||||
|
def _packages_distributions() -> dict[str, list[str]]:
|
||||||
|
try:
|
||||||
|
return dict(importlib.metadata.packages_distributions())
|
||||||
|
except Exception:
|
||||||
|
return {}
|
||||||
|
|
||||||
|
|
||||||
|
def get_module_origin(module_class: type[MVTModule]) -> ModuleOrigin:
|
||||||
|
"""Return the origin of a module class for auditing purposes."""
|
||||||
|
origin = module_class.__dict__.get(_ORIGIN_ATTRIBUTE)
|
||||||
|
if isinstance(origin, ModuleOrigin):
|
||||||
|
return origin
|
||||||
|
|
||||||
|
top_level = module_class.__module__.partition(".")[0]
|
||||||
|
if top_level == "mvt":
|
||||||
|
return ModuleOrigin(kind="builtin", name="mvt", version=MVT_VERSION)
|
||||||
|
|
||||||
|
distributions = _packages_distributions().get(top_level)
|
||||||
|
if distributions:
|
||||||
|
name = distributions[0]
|
||||||
|
version = None
|
||||||
|
commit = None
|
||||||
|
try:
|
||||||
|
dist = importlib.metadata.distribution(name)
|
||||||
|
version = dist.version
|
||||||
|
commit = _distribution_commit(dist)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
return ModuleOrigin(kind="package", name=name, version=version, commit=commit)
|
||||||
|
|
||||||
|
try:
|
||||||
|
source = str(Path(inspect.getfile(module_class)).resolve())
|
||||||
|
except (OSError, TypeError):
|
||||||
|
source = module_class.__module__
|
||||||
|
return ModuleOrigin(kind="path", name=source)
|
||||||
|
|
||||||
|
|
||||||
|
def load_installed_modules() -> list[type[MVTModule]]:
|
||||||
|
"""Load MVT modules registered by installed packages.
|
||||||
|
|
||||||
|
Packages register modules in the ``mvt.modules`` entry-point group. Each
|
||||||
|
entry point must resolve to an iterable of MVTModule subclasses, or to a
|
||||||
|
callable which returns one. A broken entry point is skipped with a
|
||||||
|
warning so that a faulty plugin package cannot break MVT.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
entry_points = importlib.metadata.entry_points(group=MODULES_ENTRY_POINT_GROUP)
|
||||||
|
except Exception as exc:
|
||||||
|
log.warning(
|
||||||
|
"Unable to discover installed module packages in entry-point group %s: %s",
|
||||||
|
MODULES_ENTRY_POINT_GROUP,
|
||||||
|
exc,
|
||||||
|
)
|
||||||
|
return []
|
||||||
|
|
||||||
|
installed_modules: list[type[MVTModule]] = []
|
||||||
|
ordered_entry_points = sorted(
|
||||||
|
entry_points, key=lambda entry_point: (entry_point.name, entry_point.value)
|
||||||
|
)
|
||||||
|
for entry_point in ordered_entry_points:
|
||||||
|
try:
|
||||||
|
loaded = entry_point.load()
|
||||||
|
if callable(loaded) and not isinstance(loaded, type):
|
||||||
|
loaded = loaded()
|
||||||
|
module_classes = list(loaded)
|
||||||
|
except (Exception, SystemExit) as exc:
|
||||||
|
log.warning(
|
||||||
|
"Unable to load modules from entry point %s (%s): %s",
|
||||||
|
entry_point.name,
|
||||||
|
entry_point.value,
|
||||||
|
exc,
|
||||||
|
)
|
||||||
|
continue
|
||||||
|
|
||||||
|
origin = _entry_point_origin(entry_point)
|
||||||
|
for module_class in module_classes:
|
||||||
|
if not (
|
||||||
|
isinstance(module_class, type) and issubclass(module_class, MVTModule)
|
||||||
|
):
|
||||||
|
log.warning(
|
||||||
|
"Entry point %s (%s) provided %r which is not an "
|
||||||
|
"MVTModule subclass",
|
||||||
|
entry_point.name,
|
||||||
|
entry_point.value,
|
||||||
|
module_class,
|
||||||
|
)
|
||||||
|
continue
|
||||||
|
setattr(module_class, _ORIGIN_ATTRIBUTE, origin)
|
||||||
|
installed_modules.append(module_class)
|
||||||
|
|
||||||
|
return installed_modules
|
||||||
|
|
||||||
|
|
||||||
def load_custom_modules(paths: Optional[Iterable[str]] = None) -> list[type[MVTModule]]:
|
def load_custom_modules(paths: Optional[Iterable[str]] = None) -> list[type[MVTModule]]:
|
||||||
search_paths: list[str] = []
|
search_paths: list[str] = []
|
||||||
env_path = os.environ.get(MVT_CUSTOM_MODULES_ENV)
|
env_path = os.environ.get(MVT_CUSTOM_MODULES_ENV)
|
||||||
@@ -105,10 +315,17 @@ def load_custom_modules(paths: Optional[Iterable[str]] = None) -> list[type[MVTM
|
|||||||
|
|
||||||
custom_modules: list[type[MVTModule]] = []
|
custom_modules: list[type[MVTModule]] = []
|
||||||
seen: set[tuple[str, str]] = set()
|
seen: set[tuple[str, str]] = set()
|
||||||
|
|
||||||
|
for module_class in load_installed_modules():
|
||||||
|
key = _module_key(module_class)
|
||||||
|
if key in seen:
|
||||||
|
continue
|
||||||
|
seen.add(key)
|
||||||
|
custom_modules.append(module_class)
|
||||||
|
|
||||||
for path in search_paths:
|
for path in search_paths:
|
||||||
for module_class in load_custom_modules_from_path(path):
|
for module_class in load_custom_modules_from_path(path):
|
||||||
source = Path(inspect.getfile(module_class)).resolve()
|
key = _module_key(module_class)
|
||||||
key = (str(source), module_class.__qualname__)
|
|
||||||
if key in seen:
|
if key in seen:
|
||||||
continue
|
continue
|
||||||
seen.add(key)
|
seen.add(key)
|
||||||
|
|||||||
@@ -0,0 +1,299 @@
|
|||||||
|
# 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 contextlib
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
import re
|
||||||
|
import tempfile
|
||||||
|
from typing import Any, ClassVar, Dict, List, Tuple, Type, TypeVar
|
||||||
|
|
||||||
|
import yaml
|
||||||
|
from appdirs import user_config_dir, user_data_dir
|
||||||
|
from pydantic import ValidationError
|
||||||
|
from pydantic_settings import (
|
||||||
|
BaseSettings,
|
||||||
|
EnvSettingsSource,
|
||||||
|
PydanticBaseSettingsSource,
|
||||||
|
SettingsConfigDict,
|
||||||
|
YamlConfigSettingsSource,
|
||||||
|
)
|
||||||
|
|
||||||
|
PLUGIN_CONFIG_FOLDER_NAME = "plugins"
|
||||||
|
# Not "plugins": on macOS the configuration and data folders are the same
|
||||||
|
# directory, and that name already holds the settings files.
|
||||||
|
PLUGIN_DATA_FOLDER_NAME = "plugin-data"
|
||||||
|
PLUGIN_ENV_PREFIX = "MVT_PLUGIN_"
|
||||||
|
|
||||||
|
PLUGIN_NAME_PATTERN = re.compile(r"[a-z0-9][a-z0-9-]*")
|
||||||
|
|
||||||
|
PluginSettingsType = TypeVar("PluginSettingsType", bound="MVTPluginSettings")
|
||||||
|
|
||||||
|
|
||||||
|
class PluginConfigLoadError(Exception):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def validate_plugin_name(plugin_name: Any) -> str:
|
||||||
|
"""
|
||||||
|
Check that a plugin name is safe to use in a file name and an env variable.
|
||||||
|
|
||||||
|
:param plugin_name: Name to validate.
|
||||||
|
:returns: The validated plugin name.
|
||||||
|
"""
|
||||||
|
if not isinstance(plugin_name, str):
|
||||||
|
raise TypeError(
|
||||||
|
f"Plugin name must be a string, not {type(plugin_name).__name__}"
|
||||||
|
)
|
||||||
|
if not PLUGIN_NAME_PATTERN.fullmatch(plugin_name):
|
||||||
|
raise ValueError(
|
||||||
|
f"Invalid plugin name {plugin_name!r}: plugin names must start with a "
|
||||||
|
"lowercase letter or a digit and may only contain lowercase letters, "
|
||||||
|
"digits and dashes"
|
||||||
|
)
|
||||||
|
return plugin_name
|
||||||
|
|
||||||
|
|
||||||
|
def plugin_config_folder() -> str:
|
||||||
|
"""
|
||||||
|
Return the folder where plugins store their configuration files.
|
||||||
|
|
||||||
|
The path is resolved on every call so it always reflects the current
|
||||||
|
environment.
|
||||||
|
"""
|
||||||
|
return os.path.join(user_config_dir("mvt"), PLUGIN_CONFIG_FOLDER_NAME)
|
||||||
|
|
||||||
|
|
||||||
|
def plugin_config_path(plugin_name: str) -> str:
|
||||||
|
"""
|
||||||
|
Return the path of the configuration file of a given plugin.
|
||||||
|
|
||||||
|
:param plugin_name: Name of the plugin.
|
||||||
|
"""
|
||||||
|
return os.path.join(
|
||||||
|
plugin_config_folder(), f"{validate_plugin_name(plugin_name)}.yaml"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def plugin_data_folder(plugin_name: str) -> str:
|
||||||
|
"""
|
||||||
|
Return the folder where a given plugin stores its data, creating it.
|
||||||
|
|
||||||
|
Plugins should keep whatever they persist, such as caches or downloaded
|
||||||
|
artifacts, in this folder. It is created with owner-only permissions. The
|
||||||
|
path is resolved on every call so it always reflects the current
|
||||||
|
environment. A plugin with a settings class calls
|
||||||
|
`MVTPluginSettings.data_folder()` instead, which passes `plugin_name` here.
|
||||||
|
|
||||||
|
:param plugin_name: Name of the plugin.
|
||||||
|
:returns: The path of the data folder of the plugin.
|
||||||
|
"""
|
||||||
|
# Validate the name before anything is created, so an unsafe name cannot
|
||||||
|
# leave a folder behind.
|
||||||
|
name = validate_plugin_name(plugin_name)
|
||||||
|
|
||||||
|
# makedirs() applies its mode only to the last folder of the path, so
|
||||||
|
# MVT's own data folder keeps the default permissions while the two
|
||||||
|
# plugin folders are private.
|
||||||
|
data_folder = os.path.join(user_data_dir("mvt"), PLUGIN_DATA_FOLDER_NAME)
|
||||||
|
os.makedirs(data_folder, mode=0o700, exist_ok=True)
|
||||||
|
|
||||||
|
folder = os.path.join(data_folder, name)
|
||||||
|
os.makedirs(folder, mode=0o700, exist_ok=True)
|
||||||
|
return folder
|
||||||
|
|
||||||
|
|
||||||
|
def plugin_env_prefix(plugin_name: str) -> str:
|
||||||
|
"""
|
||||||
|
Return the environment variable prefix used by a given plugin.
|
||||||
|
|
||||||
|
Dashes are replaced by underscores. Plugin names cannot contain underscores,
|
||||||
|
so two different plugin names never share an environment namespace.
|
||||||
|
|
||||||
|
:param plugin_name: Name of the plugin.
|
||||||
|
"""
|
||||||
|
name = validate_plugin_name(plugin_name).upper().replace("-", "_")
|
||||||
|
return f"{PLUGIN_ENV_PREFIX}{name}_"
|
||||||
|
|
||||||
|
|
||||||
|
def _settings_plugin_name(settings_cls: Type[BaseSettings]) -> str:
|
||||||
|
plugin_name = getattr(settings_cls, "plugin_name", None)
|
||||||
|
if plugin_name is None:
|
||||||
|
raise TypeError(
|
||||||
|
f"{settings_cls.__name__} must set a 'plugin_name' class attribute to "
|
||||||
|
"namespace its configuration file and environment variables"
|
||||||
|
)
|
||||||
|
return validate_plugin_name(plugin_name)
|
||||||
|
|
||||||
|
|
||||||
|
def _plugin_yaml_source(
|
||||||
|
settings_cls: Type[BaseSettings], config_path: str
|
||||||
|
) -> YamlConfigSettingsSource:
|
||||||
|
"""
|
||||||
|
Build the YAML settings source of a plugin, reporting unusable files.
|
||||||
|
|
||||||
|
A missing file is not an error, but a file which cannot be parsed or which
|
||||||
|
does not hold a mapping of setting names is reported with its path.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
return YamlConfigSettingsSource(settings_cls, config_path)
|
||||||
|
except yaml.YAMLError as exc:
|
||||||
|
raise PluginConfigLoadError(
|
||||||
|
f"Invalid plugin configuration file {config_path}: {exc}"
|
||||||
|
) from exc
|
||||||
|
except (TypeError, ValueError) as exc:
|
||||||
|
raise PluginConfigLoadError(
|
||||||
|
f"Invalid plugin configuration file {config_path}: the file must "
|
||||||
|
"contain a mapping of setting names to values"
|
||||||
|
) from exc
|
||||||
|
|
||||||
|
|
||||||
|
class MVTPluginSettings(BaseSettings):
|
||||||
|
"""
|
||||||
|
Base class for plugin-namespaced settings.
|
||||||
|
|
||||||
|
Subclass with typed fields and set `plugin_name`. Values resolve from
|
||||||
|
constructor arguments, then environment variables (MVT_PLUGIN_<NAME>_*),
|
||||||
|
then the plugin's YAML file (~/.config/mvt/plugins/<name>.yaml), then field
|
||||||
|
defaults.
|
||||||
|
|
||||||
|
Plugins must not store their settings in MVT's own configuration file: MVT
|
||||||
|
rewrites it with the fields it knows about, dropping anything else.
|
||||||
|
|
||||||
|
`data_folder()` returns the folder the plugin keeps its data in.
|
||||||
|
"""
|
||||||
|
|
||||||
|
model_config = SettingsConfigDict(extra="ignore")
|
||||||
|
|
||||||
|
plugin_name: ClassVar[str]
|
||||||
|
|
||||||
|
def __init_subclass__(cls, **kwargs: Any) -> None:
|
||||||
|
super().__init_subclass__(**kwargs)
|
||||||
|
plugin_name = _settings_plugin_name(cls)
|
||||||
|
# Namespace the environment variables of this plugin. Each pydantic
|
||||||
|
# model gets its own configuration dictionary, so this does not leak
|
||||||
|
# into other plugins.
|
||||||
|
cls.model_config["env_prefix"] = plugin_env_prefix(plugin_name)
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def settings_customise_sources(
|
||||||
|
cls,
|
||||||
|
settings_cls: Type[BaseSettings],
|
||||||
|
init_settings: PydanticBaseSettingsSource,
|
||||||
|
env_settings: PydanticBaseSettingsSource,
|
||||||
|
dotenv_settings: PydanticBaseSettingsSource,
|
||||||
|
file_secret_settings: PydanticBaseSettingsSource,
|
||||||
|
) -> Tuple[PydanticBaseSettingsSource, ...]:
|
||||||
|
config_path = plugin_config_path(_settings_plugin_name(settings_cls))
|
||||||
|
yaml_source = _plugin_yaml_source(settings_cls, config_path)
|
||||||
|
# Explicit arguments take precedence over environment variables, which
|
||||||
|
# in turn take precedence over the configuration file.
|
||||||
|
return (init_settings, env_settings, yaml_source)
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def load(cls: Type[PluginSettingsType]) -> PluginSettingsType:
|
||||||
|
"""
|
||||||
|
Load the settings of the plugin.
|
||||||
|
|
||||||
|
A missing configuration file is not an error: the settings then come
|
||||||
|
from the environment and from the field defaults.
|
||||||
|
"""
|
||||||
|
return cls()
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def data_folder(cls) -> str:
|
||||||
|
"""
|
||||||
|
Return the data folder of the plugin, creating it.
|
||||||
|
|
||||||
|
The folder is the one plugin_data_folder() returns for `plugin_name`,
|
||||||
|
so a plugin with a settings class does not repeat its name.
|
||||||
|
"""
|
||||||
|
return plugin_data_folder(_settings_plugin_name(cls))
|
||||||
|
|
||||||
|
def _environment_values(self) -> Dict[str, Any]:
|
||||||
|
"""
|
||||||
|
Return the settings values currently supplied by the environment.
|
||||||
|
|
||||||
|
Values are validated by the model, so they can be compared with the
|
||||||
|
values held by this instance.
|
||||||
|
"""
|
||||||
|
settings_cls = type(self)
|
||||||
|
raw_values = EnvSettingsSource(settings_cls)()
|
||||||
|
names = [name for name in raw_values if name in settings_cls.model_fields]
|
||||||
|
if not names:
|
||||||
|
return {}
|
||||||
|
|
||||||
|
# Fall back on the current values for the fields the environment does
|
||||||
|
# not set, so that required fields do not fail validation here.
|
||||||
|
current_values = json.loads(self.model_dump_json())
|
||||||
|
try:
|
||||||
|
from_environment = settings_cls.model_validate(
|
||||||
|
{**current_values, **raw_values}
|
||||||
|
)
|
||||||
|
except ValidationError:
|
||||||
|
# An environment variable which the model cannot validate must not
|
||||||
|
# stop the other environment values from being recognised, or a
|
||||||
|
# credential would be written to the configuration file.
|
||||||
|
return self._environment_values_by_field(current_values, raw_values, names)
|
||||||
|
return {name: getattr(from_environment, name) for name in names}
|
||||||
|
|
||||||
|
def _environment_values_by_field(
|
||||||
|
self,
|
||||||
|
current_values: Dict[str, Any],
|
||||||
|
raw_values: Dict[str, Any],
|
||||||
|
names: List[str],
|
||||||
|
) -> Dict[str, Any]:
|
||||||
|
"""
|
||||||
|
Validate each environment value on its own, skipping unusable ones.
|
||||||
|
|
||||||
|
:param current_values: Serialized values held by this instance.
|
||||||
|
:param raw_values: Values supplied by the environment.
|
||||||
|
:param names: Names of the fields set by the environment.
|
||||||
|
"""
|
||||||
|
settings_cls = type(self)
|
||||||
|
values = {}
|
||||||
|
for name in names:
|
||||||
|
try:
|
||||||
|
from_environment = settings_cls.model_validate(
|
||||||
|
{**current_values, name: raw_values[name]}
|
||||||
|
)
|
||||||
|
except ValidationError:
|
||||||
|
continue
|
||||||
|
values[name] = getattr(from_environment, name)
|
||||||
|
return values
|
||||||
|
|
||||||
|
def save(self) -> None:
|
||||||
|
"""
|
||||||
|
Save the current settings to the configuration file of the plugin.
|
||||||
|
|
||||||
|
Only values which differ from the field defaults are persisted. Values
|
||||||
|
which come from the environment are not written to disk, so credentials
|
||||||
|
passed as environment variables stay out of the configuration file.
|
||||||
|
MVT's own configuration file is never modified.
|
||||||
|
"""
|
||||||
|
config_folder = plugin_config_folder()
|
||||||
|
if not os.path.isdir(config_folder):
|
||||||
|
os.makedirs(config_folder, mode=0o700, exist_ok=True)
|
||||||
|
|
||||||
|
values = json.loads(self.model_dump_json(exclude_defaults=True))
|
||||||
|
for name, environment_value in self._environment_values().items():
|
||||||
|
if name in values and getattr(self, name, None) == environment_value:
|
||||||
|
del values[name]
|
||||||
|
|
||||||
|
# Settings files can hold credentials, so write them through a private
|
||||||
|
# temporary file and move it in place. The file is then never partially
|
||||||
|
# written and never briefly readable by other users.
|
||||||
|
config_path = plugin_config_path(self.plugin_name)
|
||||||
|
descriptor, temporary_path = tempfile.mkstemp(
|
||||||
|
dir=config_folder, prefix=f".{self.plugin_name}-", suffix=".yaml"
|
||||||
|
)
|
||||||
|
try:
|
||||||
|
with os.fdopen(descriptor, "w") as config_file:
|
||||||
|
config_file.write(yaml.dump(values, default_flow_style=False))
|
||||||
|
os.replace(temporary_path, config_path)
|
||||||
|
except BaseException:
|
||||||
|
with contextlib.suppress(OSError):
|
||||||
|
os.unlink(temporary_path)
|
||||||
|
raise
|
||||||
@@ -239,6 +239,13 @@ def init_logging(verbose: bool = False):
|
|||||||
"""
|
"""
|
||||||
log = logging.getLogger("mvt")
|
log = logging.getLogger("mvt")
|
||||||
log.setLevel(logging.DEBUG)
|
log.setLevel(logging.DEBUG)
|
||||||
|
|
||||||
|
# Importing an MVT CLI module calls init_logging() at import time, and
|
||||||
|
# loaded module packages may import one indirectly. Keep this idempotent
|
||||||
|
# so console log lines are not duplicated by a second handler.
|
||||||
|
if any(isinstance(handler, MVTLogHandler) for handler in log.handlers):
|
||||||
|
return
|
||||||
|
|
||||||
consoleHandler = MVTLogHandler()
|
consoleHandler = MVTLogHandler()
|
||||||
consoleHandler.setFormatter(logging.Formatter("%(message)s"))
|
consoleHandler.setFormatter(logging.Formatter("%(message)s"))
|
||||||
if verbose:
|
if verbose:
|
||||||
|
|||||||
@@ -157,7 +157,7 @@ class TestCommand:
|
|||||||
assert not hasattr(cmd, "initialized")
|
assert not hasattr(cmd, "initialized")
|
||||||
assert "Circular module dependency detected" in caplog.text
|
assert "Circular module dependency detected" in caplog.text
|
||||||
|
|
||||||
def test_unavailable_dependency_warns_and_stops(self, caplog):
|
def test_unavailable_dependency_only_skips_the_dependent_module(self, caplog):
|
||||||
class UnavailableModule(RecordingModule):
|
class UnavailableModule(RecordingModule):
|
||||||
pass
|
pass
|
||||||
|
|
||||||
@@ -165,14 +165,97 @@ class TestCommand:
|
|||||||
dependencies = (UnavailableModule,)
|
dependencies = (UnavailableModule,)
|
||||||
|
|
||||||
cmd = RecordingCommand()
|
cmd = RecordingCommand()
|
||||||
cmd.modules = [DependentModule]
|
cmd.modules = [DependentModule, IndependentModule, FirstModule]
|
||||||
|
|
||||||
|
with caplog.at_level(logging.WARNING):
|
||||||
|
cmd.run()
|
||||||
|
|
||||||
|
assert RecordingModule.run_order == ["IndependentModule", "FirstModule"]
|
||||||
|
assert cmd.initialized
|
||||||
|
assert "Module DependentModule will be SKIPPED" in caplog.text
|
||||||
|
assert "depends on module UnavailableModule" in caplog.text
|
||||||
|
|
||||||
|
def test_modules_depending_on_a_skipped_module_are_skipped_too(self, caplog):
|
||||||
|
class UnavailableModule(RecordingModule):
|
||||||
|
pass
|
||||||
|
|
||||||
|
class SkippedModule(RecordingModule):
|
||||||
|
dependencies = (UnavailableModule,)
|
||||||
|
|
||||||
|
class DependsOnSkippedModule(RecordingModule):
|
||||||
|
dependencies = (SkippedModule,)
|
||||||
|
|
||||||
|
class DependsOnTheChain(RecordingModule):
|
||||||
|
dependencies = (DependsOnSkippedModule,)
|
||||||
|
|
||||||
|
cmd = RecordingCommand()
|
||||||
|
cmd.modules = [
|
||||||
|
DependsOnTheChain,
|
||||||
|
DependsOnSkippedModule,
|
||||||
|
SkippedModule,
|
||||||
|
IndependentModule,
|
||||||
|
]
|
||||||
|
|
||||||
|
with caplog.at_level(logging.WARNING):
|
||||||
|
cmd.run()
|
||||||
|
|
||||||
|
assert RecordingModule.run_order == ["IndependentModule"]
|
||||||
|
skip_warnings = [
|
||||||
|
record.getMessage()
|
||||||
|
for record in caplog.records
|
||||||
|
if "will be SKIPPED" in record.getMessage()
|
||||||
|
]
|
||||||
|
assert len(skip_warnings) == 3
|
||||||
|
assert [warning.split()[1] for warning in skip_warnings] == [
|
||||||
|
"SkippedModule",
|
||||||
|
"DependsOnSkippedModule",
|
||||||
|
"DependsOnTheChain",
|
||||||
|
]
|
||||||
|
# Every warning names the root cause: the module missing a dependency
|
||||||
|
# and the dependency it is missing.
|
||||||
|
assert all("UnavailableModule" in warning for warning in skip_warnings)
|
||||||
|
assert all("module SkippedModule" in warning for warning in skip_warnings[1:])
|
||||||
|
|
||||||
|
def test_explicitly_selected_module_with_missing_dependency_runs_nothing(
|
||||||
|
self, caplog
|
||||||
|
):
|
||||||
|
class UnavailableModule(RecordingModule):
|
||||||
|
pass
|
||||||
|
|
||||||
|
class DependentModule(RecordingModule):
|
||||||
|
dependencies = (UnavailableModule,)
|
||||||
|
|
||||||
|
cmd = RecordingCommand(module_name="DependentModule")
|
||||||
|
cmd.modules = [DependentModule, IndependentModule]
|
||||||
|
|
||||||
with caplog.at_level(logging.WARNING):
|
with caplog.at_level(logging.WARNING):
|
||||||
cmd.run()
|
cmd.run()
|
||||||
|
|
||||||
assert RecordingModule.run_order == []
|
assert RecordingModule.run_order == []
|
||||||
assert not hasattr(cmd, "initialized")
|
assert "Module DependentModule will be SKIPPED" in caplog.text
|
||||||
assert "depends on unavailable module UnavailableModule" in caplog.text
|
assert "No modules will be run" in caplog.text
|
||||||
|
# Nothing else was selected, so the warning 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
|
||||||
|
|
||||||
|
def test_unaffected_dependency_chains_keep_their_order(self, caplog):
|
||||||
|
class UnavailableModule(RecordingModule):
|
||||||
|
pass
|
||||||
|
|
||||||
|
class SkippedModule(RecordingModule):
|
||||||
|
dependencies = (UnavailableModule,)
|
||||||
|
|
||||||
|
cmd = RecordingCommand()
|
||||||
|
cmd.modules = [ThirdModule, SkippedModule, SecondModule, FirstModule]
|
||||||
|
|
||||||
|
with caplog.at_level(logging.WARNING):
|
||||||
|
cmd.run()
|
||||||
|
|
||||||
|
assert RecordingModule.run_order == [
|
||||||
|
"FirstModule",
|
||||||
|
"SecondModule",
|
||||||
|
"ThirdModule",
|
||||||
|
]
|
||||||
|
|
||||||
def test_custom_modules_are_filtered_before_ordering(self):
|
def test_custom_modules_are_filtered_before_ordering(self):
|
||||||
cmd = RecordingCommand()
|
cmd = RecordingCommand()
|
||||||
|
|||||||
@@ -3,10 +3,12 @@ import pytest
|
|||||||
from mvt.common.module import MVTModule
|
from mvt.common.module import MVTModule
|
||||||
from mvt.common.module_loader import (
|
from mvt.common.module_loader import (
|
||||||
CustomModuleLoadError,
|
CustomModuleLoadError,
|
||||||
|
get_module_logger,
|
||||||
load_custom_modules,
|
load_custom_modules,
|
||||||
load_custom_modules_from_path,
|
load_custom_modules_from_path,
|
||||||
module_supports_command,
|
module_supports_command,
|
||||||
)
|
)
|
||||||
|
from mvt.ios.modules.mixed.whatsapp import Whatsapp
|
||||||
|
|
||||||
|
|
||||||
MODULE_TEMPLATE = """
|
MODULE_TEMPLATE = """
|
||||||
@@ -144,3 +146,44 @@ def test_module_supports_command_honors_supported_commands(tmp_path):
|
|||||||
|
|
||||||
assert module_supports_command(module, "ios", "check-backup")
|
assert module_supports_command(module, "ios", "check-backup")
|
||||||
assert not module_supports_command(module, "ios", "check-fs")
|
assert not module_supports_command(module, "ios", "check-fs")
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_module_logger_keeps_builtin_names():
|
||||||
|
assert get_module_logger(Whatsapp).name == "mvt.ios.modules.mixed.whatsapp"
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_module_logger_parents_package_modules_under_mvt_ext():
|
||||||
|
class PackageModule(MVTModule):
|
||||||
|
pass
|
||||||
|
|
||||||
|
PackageModule.__module__ = "some_plugin_package.ios.custom"
|
||||||
|
|
||||||
|
assert (
|
||||||
|
get_module_logger(PackageModule).name
|
||||||
|
== "mvt.ext.some_plugin_package.ios.custom"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_module_logger_strips_the_plugin_package_prefix():
|
||||||
|
class PluginModule(MVTModule):
|
||||||
|
pass
|
||||||
|
|
||||||
|
PluginModule.__module__ = "mvt_plugin_amnesty_custom.ios.custom"
|
||||||
|
|
||||||
|
assert get_module_logger(PluginModule).name == "mvt.ext.amnesty_custom.ios.custom"
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_module_logger_only_strips_the_prefix_from_the_top_level():
|
||||||
|
class NestedModule(MVTModule):
|
||||||
|
pass
|
||||||
|
|
||||||
|
NestedModule.__module__ = "other_package.mvt_plugin_sub"
|
||||||
|
|
||||||
|
assert get_module_logger(NestedModule).name == "mvt.ext.other_package.mvt_plugin_sub"
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_module_logger_names_path_modules_after_their_file(tmp_path):
|
||||||
|
module_path = _write_module(tmp_path / "my_custom_module.py", "PathModule")
|
||||||
|
module = load_custom_modules_from_path(str(module_path))[0]
|
||||||
|
|
||||||
|
assert get_module_logger(module).name == "mvt.ext.my_custom_module"
|
||||||
@@ -0,0 +1,391 @@
|
|||||||
|
# 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 os
|
||||||
|
import stat
|
||||||
|
import sys
|
||||||
|
from typing import Optional
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
import yaml
|
||||||
|
|
||||||
|
from mvt.common.plugin_config import (
|
||||||
|
MVTPluginSettings,
|
||||||
|
PluginConfigLoadError,
|
||||||
|
plugin_config_folder,
|
||||||
|
plugin_config_path,
|
||||||
|
plugin_data_folder,
|
||||||
|
plugin_env_prefix,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class ExamplePluginSettings(MVTPluginSettings):
|
||||||
|
plugin_name = "example-plugin"
|
||||||
|
|
||||||
|
API_KEY: Optional[str] = None
|
||||||
|
CACHE_FOLDER: str = "cache"
|
||||||
|
MAX_RESULTS: int = 25
|
||||||
|
|
||||||
|
|
||||||
|
class OtherPluginSettings(MVTPluginSettings):
|
||||||
|
plugin_name = "other-plugin"
|
||||||
|
|
||||||
|
API_KEY: Optional[str] = None
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def config_folder(tmp_path, monkeypatch):
|
||||||
|
monkeypatch.setattr(
|
||||||
|
"mvt.common.plugin_config.user_config_dir",
|
||||||
|
lambda *args, **kwargs: str(tmp_path),
|
||||||
|
)
|
||||||
|
return tmp_path
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def data_folder(tmp_path, monkeypatch):
|
||||||
|
folder = tmp_path / "data"
|
||||||
|
monkeypatch.setattr(
|
||||||
|
"mvt.common.plugin_config.user_data_dir",
|
||||||
|
lambda *args, **kwargs: str(folder),
|
||||||
|
)
|
||||||
|
return folder
|
||||||
|
|
||||||
|
|
||||||
|
def _write_plugin_file(plugin_name, values):
|
||||||
|
config_path = plugin_config_path(plugin_name)
|
||||||
|
os.makedirs(os.path.dirname(config_path), exist_ok=True)
|
||||||
|
content = values if isinstance(values, str) else yaml.dump(values)
|
||||||
|
with open(config_path, "w") as config_file:
|
||||||
|
config_file.write(content)
|
||||||
|
return config_path
|
||||||
|
|
||||||
|
|
||||||
|
def test_plugin_paths_and_prefixes_are_namespaced(config_folder):
|
||||||
|
assert plugin_config_folder() == str(config_folder / "plugins")
|
||||||
|
assert plugin_config_path("example-plugin") == str(
|
||||||
|
config_folder / "plugins" / "example-plugin.yaml"
|
||||||
|
)
|
||||||
|
assert plugin_env_prefix("example-plugin") == "MVT_PLUGIN_EXAMPLE_PLUGIN_"
|
||||||
|
assert plugin_env_prefix("other-plugin") == "MVT_PLUGIN_OTHER_PLUGIN_"
|
||||||
|
|
||||||
|
|
||||||
|
def test_defaults_are_used_without_file_or_environment(config_folder):
|
||||||
|
settings = ExamplePluginSettings.load()
|
||||||
|
|
||||||
|
assert settings.API_KEY is None
|
||||||
|
assert settings.CACHE_FOLDER == "cache"
|
||||||
|
assert settings.MAX_RESULTS == 25
|
||||||
|
assert not os.path.exists(plugin_config_path("example-plugin"))
|
||||||
|
|
||||||
|
|
||||||
|
def test_values_are_loaded_from_the_plugin_file(config_folder):
|
||||||
|
_write_plugin_file("example-plugin", {"API_KEY": "from-file", "MAX_RESULTS": 5})
|
||||||
|
|
||||||
|
settings = ExamplePluginSettings.load()
|
||||||
|
|
||||||
|
assert settings.API_KEY == "from-file"
|
||||||
|
assert settings.MAX_RESULTS == 5
|
||||||
|
assert settings.CACHE_FOLDER == "cache"
|
||||||
|
|
||||||
|
|
||||||
|
def test_environment_overrides_the_plugin_file(config_folder, monkeypatch):
|
||||||
|
_write_plugin_file("example-plugin", {"API_KEY": "from-file", "MAX_RESULTS": 5})
|
||||||
|
monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "from-environment")
|
||||||
|
|
||||||
|
settings = ExamplePluginSettings.load()
|
||||||
|
|
||||||
|
assert settings.API_KEY == "from-environment"
|
||||||
|
assert settings.MAX_RESULTS == 5
|
||||||
|
|
||||||
|
|
||||||
|
def test_arguments_override_the_environment_and_the_plugin_file(
|
||||||
|
config_folder, monkeypatch
|
||||||
|
):
|
||||||
|
_write_plugin_file("example-plugin", {"API_KEY": "from-file", "MAX_RESULTS": 5})
|
||||||
|
monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "from-environment")
|
||||||
|
monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_MAX_RESULTS", "10")
|
||||||
|
|
||||||
|
settings = ExamplePluginSettings(API_KEY="from-argument")
|
||||||
|
|
||||||
|
assert settings.API_KEY == "from-argument"
|
||||||
|
assert settings.MAX_RESULTS == 10
|
||||||
|
|
||||||
|
|
||||||
|
def test_save_and_load_round_trip(config_folder):
|
||||||
|
settings = ExamplePluginSettings.load()
|
||||||
|
settings.API_KEY = "saved-key"
|
||||||
|
settings.MAX_RESULTS = 100
|
||||||
|
|
||||||
|
settings.save()
|
||||||
|
|
||||||
|
config_path = plugin_config_path("example-plugin")
|
||||||
|
assert os.path.isfile(config_path)
|
||||||
|
with open(config_path) as config_file:
|
||||||
|
assert yaml.safe_load(config_file) == {
|
||||||
|
"API_KEY": "saved-key",
|
||||||
|
"MAX_RESULTS": 100,
|
||||||
|
}
|
||||||
|
|
||||||
|
reloaded = ExamplePluginSettings.load()
|
||||||
|
assert reloaded.API_KEY == "saved-key"
|
||||||
|
assert reloaded.MAX_RESULTS == 100
|
||||||
|
assert reloaded.CACHE_FOLDER == "cache"
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.skipif(
|
||||||
|
sys.platform == "win32", reason="POSIX file permissions are not available"
|
||||||
|
)
|
||||||
|
def test_saved_file_is_only_readable_by_the_user(config_folder):
|
||||||
|
settings = ExamplePluginSettings.load()
|
||||||
|
settings.API_KEY = "saved-key"
|
||||||
|
|
||||||
|
settings.save()
|
||||||
|
|
||||||
|
config_path = plugin_config_path("example-plugin")
|
||||||
|
assert stat.S_IMODE(os.stat(config_path).st_mode) == 0o600
|
||||||
|
folder_mode = stat.S_IMODE(os.stat(plugin_config_folder()).st_mode)
|
||||||
|
assert folder_mode & 0o077 == 0
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.skipif(
|
||||||
|
sys.platform == "win32", reason="POSIX file permissions are not available"
|
||||||
|
)
|
||||||
|
def test_save_restricts_the_permissions_of_an_existing_file(config_folder):
|
||||||
|
config_path = _write_plugin_file("example-plugin", {"API_KEY": "from-file"})
|
||||||
|
os.chmod(config_path, 0o644)
|
||||||
|
|
||||||
|
settings = ExamplePluginSettings.load()
|
||||||
|
settings.MAX_RESULTS = 100
|
||||||
|
settings.save()
|
||||||
|
|
||||||
|
assert stat.S_IMODE(os.stat(config_path).st_mode) == 0o600
|
||||||
|
assert os.listdir(plugin_config_folder()) == ["example-plugin.yaml"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_save_only_persists_non_default_values(config_folder):
|
||||||
|
settings = ExamplePluginSettings.load()
|
||||||
|
settings.CACHE_FOLDER = "another-cache"
|
||||||
|
|
||||||
|
settings.save()
|
||||||
|
|
||||||
|
with open(plugin_config_path("example-plugin")) as config_file:
|
||||||
|
assert yaml.safe_load(config_file) == {"CACHE_FOLDER": "another-cache"}
|
||||||
|
|
||||||
|
|
||||||
|
def test_save_does_not_persist_values_coming_from_the_environment(
|
||||||
|
config_folder, monkeypatch
|
||||||
|
):
|
||||||
|
monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "environment-secret")
|
||||||
|
|
||||||
|
settings = ExamplePluginSettings.load()
|
||||||
|
assert settings.API_KEY == "environment-secret"
|
||||||
|
settings.MAX_RESULTS = 100
|
||||||
|
settings.save()
|
||||||
|
|
||||||
|
with open(plugin_config_path("example-plugin")) as config_file:
|
||||||
|
assert yaml.safe_load(config_file) == {"MAX_RESULTS": 100}
|
||||||
|
|
||||||
|
|
||||||
|
def test_an_invalid_environment_variable_still_protects_the_other_values(
|
||||||
|
config_folder, monkeypatch
|
||||||
|
):
|
||||||
|
monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "environment-secret")
|
||||||
|
monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_MAX_RESULTS", "not-an-int")
|
||||||
|
|
||||||
|
settings = ExamplePluginSettings(MAX_RESULTS=100)
|
||||||
|
assert settings.API_KEY == "environment-secret"
|
||||||
|
settings.save()
|
||||||
|
|
||||||
|
with open(plugin_config_path("example-plugin")) as config_file:
|
||||||
|
saved_values = yaml.safe_load(config_file)
|
||||||
|
assert saved_values == {"MAX_RESULTS": 100}
|
||||||
|
assert "API_KEY" not in saved_values
|
||||||
|
|
||||||
|
|
||||||
|
def test_save_persists_values_which_differ_from_the_environment(
|
||||||
|
config_folder, monkeypatch
|
||||||
|
):
|
||||||
|
monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "environment-secret")
|
||||||
|
|
||||||
|
settings = ExamplePluginSettings.load()
|
||||||
|
settings.API_KEY = "chosen-key"
|
||||||
|
settings.save()
|
||||||
|
|
||||||
|
with open(plugin_config_path("example-plugin")) as config_file:
|
||||||
|
assert yaml.safe_load(config_file) == {"API_KEY": "chosen-key"}
|
||||||
|
|
||||||
|
|
||||||
|
def test_save_does_not_write_the_mvt_configuration_file(config_folder):
|
||||||
|
settings = ExamplePluginSettings.load()
|
||||||
|
settings.API_KEY = "saved-key"
|
||||||
|
|
||||||
|
settings.save()
|
||||||
|
|
||||||
|
assert os.listdir(config_folder) == ["plugins"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_unknown_keys_in_the_plugin_file_are_ignored(config_folder):
|
||||||
|
_write_plugin_file(
|
||||||
|
"example-plugin",
|
||||||
|
{"API_KEY": "from-file", "UNKNOWN_SETTING": "ignored"},
|
||||||
|
)
|
||||||
|
|
||||||
|
settings = ExamplePluginSettings.load()
|
||||||
|
|
||||||
|
assert settings.API_KEY == "from-file"
|
||||||
|
assert not hasattr(settings, "UNKNOWN_SETTING")
|
||||||
|
|
||||||
|
|
||||||
|
def test_unparsable_plugin_file_is_reported_with_its_path(config_folder):
|
||||||
|
config_path = _write_plugin_file("example-plugin", "API_KEY: [unclosed\n")
|
||||||
|
|
||||||
|
with pytest.raises(PluginConfigLoadError) as raised:
|
||||||
|
ExamplePluginSettings.load()
|
||||||
|
|
||||||
|
assert config_path in str(raised.value)
|
||||||
|
|
||||||
|
|
||||||
|
def test_plugin_file_which_is_not_a_mapping_is_reported_with_its_path(config_folder):
|
||||||
|
config_path = _write_plugin_file("example-plugin", "- one\n- two\n")
|
||||||
|
|
||||||
|
with pytest.raises(PluginConfigLoadError) as raised:
|
||||||
|
ExamplePluginSettings.load()
|
||||||
|
|
||||||
|
assert config_path in str(raised.value)
|
||||||
|
assert "mapping of setting names" in str(raised.value)
|
||||||
|
|
||||||
|
|
||||||
|
def test_plugins_do_not_interfere_with_each_other(config_folder, monkeypatch):
|
||||||
|
_write_plugin_file("other-plugin", {"API_KEY": "other-file-key"})
|
||||||
|
monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "example-environment-key")
|
||||||
|
|
||||||
|
example_settings = ExamplePluginSettings.load()
|
||||||
|
other_settings = OtherPluginSettings.load()
|
||||||
|
|
||||||
|
assert example_settings.API_KEY == "example-environment-key"
|
||||||
|
assert other_settings.API_KEY == "other-file-key"
|
||||||
|
|
||||||
|
example_settings.MAX_RESULTS = 100
|
||||||
|
example_settings.save()
|
||||||
|
assert sorted(os.listdir(plugin_config_folder())) == [
|
||||||
|
"example-plugin.yaml",
|
||||||
|
"other-plugin.yaml",
|
||||||
|
]
|
||||||
|
with open(plugin_config_path("other-plugin")) as config_file:
|
||||||
|
assert yaml.safe_load(config_file) == {"API_KEY": "other-file-key"}
|
||||||
|
|
||||||
|
|
||||||
|
def test_subclass_without_plugin_name_is_rejected():
|
||||||
|
with pytest.raises(TypeError, match="plugin_name"):
|
||||||
|
|
||||||
|
class MissingNameSettings(MVTPluginSettings):
|
||||||
|
API_KEY: Optional[str] = None
|
||||||
|
|
||||||
|
|
||||||
|
def test_subclass_with_invalid_plugin_name_is_rejected():
|
||||||
|
with pytest.raises(ValueError, match="Invalid plugin name"):
|
||||||
|
|
||||||
|
class InvalidNameSettings(MVTPluginSettings):
|
||||||
|
plugin_name = "Bad/Name"
|
||||||
|
|
||||||
|
|
||||||
|
def test_underscores_are_not_allowed_in_plugin_names():
|
||||||
|
# Underscores are replaced by dashes in the environment prefix, so allowing
|
||||||
|
# both would let two plugin names share one environment namespace.
|
||||||
|
with pytest.raises(ValueError, match="Invalid plugin name"):
|
||||||
|
|
||||||
|
class UnderscoreNameSettings(MVTPluginSettings):
|
||||||
|
plugin_name = "under_score"
|
||||||
|
|
||||||
|
with pytest.raises(ValueError, match="Invalid plugin name"):
|
||||||
|
plugin_config_path("under_score")
|
||||||
|
with pytest.raises(ValueError, match="Invalid plugin name"):
|
||||||
|
plugin_env_prefix("under_score")
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
"plugin_name", ["../escape", "folder/name", "UPPER", "-dash", ""]
|
||||||
|
)
|
||||||
|
def test_unsafe_plugin_names_have_no_configuration_path(plugin_name):
|
||||||
|
with pytest.raises(ValueError, match="Invalid plugin name"):
|
||||||
|
plugin_config_path(plugin_name)
|
||||||
|
|
||||||
|
|
||||||
|
def test_data_folder_is_namespaced_and_created(data_folder):
|
||||||
|
folder = plugin_data_folder("example-plugin")
|
||||||
|
|
||||||
|
assert folder == str(data_folder / "plugin-data" / "example-plugin")
|
||||||
|
assert os.path.isdir(folder)
|
||||||
|
|
||||||
|
|
||||||
|
def test_data_folder_can_be_requested_repeatedly(data_folder):
|
||||||
|
folder = plugin_data_folder("example-plugin")
|
||||||
|
with open(os.path.join(folder, "kept.json"), "w") as data_file:
|
||||||
|
data_file.write("{}")
|
||||||
|
|
||||||
|
assert plugin_data_folder("example-plugin") == folder
|
||||||
|
assert os.listdir(folder) == ["kept.json"]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.skipif(
|
||||||
|
sys.platform == "win32", reason="POSIX file permissions are not available"
|
||||||
|
)
|
||||||
|
def test_data_folder_is_only_accessible_by_the_user(data_folder):
|
||||||
|
folder = plugin_data_folder("example-plugin")
|
||||||
|
|
||||||
|
assert stat.S_IMODE(os.stat(folder).st_mode) & 0o077 == 0
|
||||||
|
parent_mode = stat.S_IMODE(os.stat(os.path.dirname(folder)).st_mode)
|
||||||
|
assert parent_mode & 0o077 == 0
|
||||||
|
|
||||||
|
|
||||||
|
def test_plugins_get_their_own_data_folder(data_folder):
|
||||||
|
example_folder = plugin_data_folder("example-plugin")
|
||||||
|
other_folder = plugin_data_folder("other-plugin")
|
||||||
|
|
||||||
|
assert example_folder != other_folder
|
||||||
|
assert sorted(os.listdir(data_folder / "plugin-data")) == [
|
||||||
|
"example-plugin",
|
||||||
|
"other-plugin",
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
def test_data_folder_does_not_touch_the_configuration_folder(
|
||||||
|
config_folder, data_folder
|
||||||
|
):
|
||||||
|
plugin_data_folder("example-plugin")
|
||||||
|
|
||||||
|
assert not os.path.exists(plugin_config_folder())
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
"plugin_name", ["../escape", "folder/name", "UPPER", "-dash", ""]
|
||||||
|
)
|
||||||
|
def test_unsafe_plugin_names_have_no_data_folder(data_folder, plugin_name):
|
||||||
|
with pytest.raises(ValueError, match="Invalid plugin name"):
|
||||||
|
plugin_data_folder(plugin_name)
|
||||||
|
|
||||||
|
assert not os.path.exists(data_folder)
|
||||||
|
|
||||||
|
|
||||||
|
def test_settings_class_knows_its_data_folder(data_folder):
|
||||||
|
folder = ExamplePluginSettings.data_folder()
|
||||||
|
|
||||||
|
assert folder == plugin_data_folder("example-plugin")
|
||||||
|
assert os.path.isdir(folder)
|
||||||
|
assert OtherPluginSettings.data_folder() != folder
|
||||||
|
|
||||||
|
|
||||||
|
def test_settings_instance_uses_the_same_data_folder(config_folder, data_folder):
|
||||||
|
settings = ExamplePluginSettings.load()
|
||||||
|
|
||||||
|
assert settings.data_folder() == ExamplePluginSettings.data_folder()
|
||||||
|
|
||||||
|
|
||||||
|
def test_subclass_without_its_own_name_shares_the_data_folder(data_folder):
|
||||||
|
class InheritingSettings(ExamplePluginSettings):
|
||||||
|
pass
|
||||||
|
|
||||||
|
assert InheritingSettings.data_folder() == ExamplePluginSettings.data_folder()
|
||||||
@@ -8,6 +8,7 @@ import logging
|
|||||||
import os
|
import os
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
|
|
||||||
|
from mvt.common.log import MVTLogHandler
|
||||||
from mvt.common.utils import (
|
from mvt.common.utils import (
|
||||||
CustomJSONEncoder,
|
CustomJSONEncoder,
|
||||||
convert_datetime_to_iso,
|
convert_datetime_to_iso,
|
||||||
@@ -16,6 +17,7 @@ from mvt.common.utils import (
|
|||||||
convert_unix_to_utc_datetime,
|
convert_unix_to_utc_datetime,
|
||||||
generate_hashes_from_path,
|
generate_hashes_from_path,
|
||||||
get_sha256_from_file_path,
|
get_sha256_from_file_path,
|
||||||
|
init_logging,
|
||||||
)
|
)
|
||||||
|
|
||||||
from ..utils import get_artifact_folder
|
from ..utils import get_artifact_folder
|
||||||
@@ -103,3 +105,20 @@ class TestCustomJSONEncoder:
|
|||||||
json.dumps({"name": "家".encode()}, cls=CustomJSONEncoder)
|
json.dumps({"name": "家".encode()}, cls=CustomJSONEncoder)
|
||||||
== '{"name": "\\u5bb6"}'
|
== '{"name": "\\u5bb6"}'
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class TestInitLogging:
|
||||||
|
def test__init_logging_is_idempotent(self):
|
||||||
|
# Loaded module packages may import an MVT CLI module, which calls
|
||||||
|
# init_logging() again at import time. A second call must not add
|
||||||
|
# a duplicate console handler.
|
||||||
|
log = logging.getLogger("mvt")
|
||||||
|
init_logging()
|
||||||
|
handler_count = sum(
|
||||||
|
isinstance(handler, MVTLogHandler) for handler in log.handlers
|
||||||
|
)
|
||||||
|
init_logging()
|
||||||
|
assert (
|
||||||
|
sum(isinstance(handler, MVTLogHandler) for handler in log.handlers)
|
||||||
|
== handler_count
|
||||||
|
)
|
||||||
@@ -1,3 +1,7 @@
|
|||||||
|
import hashlib
|
||||||
|
import importlib.metadata
|
||||||
|
import json
|
||||||
|
|
||||||
from click.testing import CliRunner
|
from click.testing import CliRunner
|
||||||
|
|
||||||
from mvt.android.cli import check_bugreport
|
from mvt.android.cli import check_bugreport
|
||||||
@@ -5,7 +9,9 @@ from mvt.android.cmd_check_androidqf import CmdAndroidCheckAndroidQF
|
|||||||
from mvt.android.cmd_check_backup import CmdAndroidCheckBackup
|
from mvt.android.cmd_check_backup import CmdAndroidCheckBackup
|
||||||
from mvt.android.cmd_check_bugreport import CmdAndroidCheckBugreport
|
from mvt.android.cmd_check_bugreport import CmdAndroidCheckBugreport
|
||||||
from mvt.android.cmd_check_intrusion_logs import CmdAndroidCheckIntrusionLogs
|
from mvt.android.cmd_check_intrusion_logs import CmdAndroidCheckIntrusionLogs
|
||||||
|
from mvt.common import module_loader
|
||||||
from mvt.common.module import MVTModule
|
from mvt.common.module import MVTModule
|
||||||
|
from mvt.common.version import MVT_VERSION
|
||||||
from mvt.ios.cli import check_backup, check_fs
|
from mvt.ios.cli import check_backup, check_fs
|
||||||
|
|
||||||
|
|
||||||
@@ -106,6 +112,150 @@ def test_custom_modules_load_from_environment_without_cli_flag(tmp_path, monkeyp
|
|||||||
assert "EnvBugreportModule" in result.output
|
assert "EnvBugreportModule" in result.output
|
||||||
|
|
||||||
|
|
||||||
|
class InstalledPackageModule(MVTModule):
|
||||||
|
supported_commands = (("ios", "check-backup"),)
|
||||||
|
|
||||||
|
|
||||||
|
def get_installed_package_modules():
|
||||||
|
return [InstalledPackageModule]
|
||||||
|
|
||||||
|
|
||||||
|
def _fake_entry_points(monkeypatch, value, name="test-modules"):
|
||||||
|
entry_point = importlib.metadata.EntryPoint(
|
||||||
|
name=name, value=value, group=module_loader.MODULES_ENTRY_POINT_GROUP
|
||||||
|
)
|
||||||
|
|
||||||
|
def fake_entry_points(*, group):
|
||||||
|
assert group == module_loader.MODULES_ENTRY_POINT_GROUP
|
||||||
|
return [entry_point]
|
||||||
|
|
||||||
|
monkeypatch.setattr(
|
||||||
|
module_loader.importlib.metadata, "entry_points", fake_entry_points
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_installed_module_package_loads_from_entry_point(monkeypatch):
|
||||||
|
_fake_entry_points(monkeypatch, f"{__name__}:get_installed_package_modules")
|
||||||
|
|
||||||
|
modules = module_loader.load_custom_modules()
|
||||||
|
|
||||||
|
assert modules == [InstalledPackageModule]
|
||||||
|
|
||||||
|
|
||||||
|
def test_broken_module_entry_point_is_skipped(monkeypatch, caplog):
|
||||||
|
_fake_entry_points(monkeypatch, "nonexistent_module_xyz:get_modules")
|
||||||
|
|
||||||
|
with caplog.at_level("WARNING"):
|
||||||
|
modules = module_loader.load_custom_modules()
|
||||||
|
|
||||||
|
assert modules == []
|
||||||
|
assert "Unable to load modules from entry point" in caplog.text
|
||||||
|
|
||||||
|
|
||||||
|
def test_entry_point_module_deduplicated_against_paths(monkeypatch, tmp_path):
|
||||||
|
_fake_entry_points(monkeypatch, f"{__name__}:get_installed_package_modules")
|
||||||
|
module_path = _write_custom_module(
|
||||||
|
tmp_path / "custom.py",
|
||||||
|
"PathLoadedModule",
|
||||||
|
(("ios", "check-backup"),),
|
||||||
|
)
|
||||||
|
|
||||||
|
modules = module_loader.load_custom_modules([str(module_path)])
|
||||||
|
|
||||||
|
assert [module.__name__ for module in modules] == [
|
||||||
|
"InstalledPackageModule",
|
||||||
|
"PathLoadedModule",
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
def test_list_modules_shows_module_sources(tmp_path, caplog):
|
||||||
|
module_path = _write_custom_module(
|
||||||
|
tmp_path / "custom.py",
|
||||||
|
"SourcedBackupModule",
|
||||||
|
(("ios", "check-backup"),),
|
||||||
|
)
|
||||||
|
file_sha256 = hashlib.sha256(module_path.read_bytes()).hexdigest()
|
||||||
|
custom_modules = module_loader.load_custom_modules([str(module_path)])
|
||||||
|
|
||||||
|
from mvt.ios.cmd_check_backup import CmdIOSCheckBackup
|
||||||
|
|
||||||
|
cmd = CmdIOSCheckBackup(target_path=str(tmp_path), custom_modules=custom_modules)
|
||||||
|
cmd.list_modules()
|
||||||
|
|
||||||
|
assert f" - Modules from 'mvt@{MVT_VERSION}':" in caplog.text
|
||||||
|
assert (
|
||||||
|
f" - Modules from '{module_path}' (sha256: {file_sha256}): SourcedBackupModule"
|
||||||
|
in caplog.text
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_builtin_module_origin():
|
||||||
|
from mvt.ios.modules.backup import BACKUP_MODULES
|
||||||
|
|
||||||
|
origin = module_loader.get_module_origin(BACKUP_MODULES[0])
|
||||||
|
|
||||||
|
assert origin.kind == "builtin"
|
||||||
|
assert origin.name == "mvt"
|
||||||
|
assert origin.version == MVT_VERSION
|
||||||
|
|
||||||
|
|
||||||
|
def test_installed_module_origin(monkeypatch):
|
||||||
|
_fake_entry_points(monkeypatch, f"{__name__}:get_installed_package_modules")
|
||||||
|
|
||||||
|
modules = module_loader.load_custom_modules()
|
||||||
|
|
||||||
|
origin = module_loader.get_module_origin(modules[0])
|
||||||
|
assert origin.kind == "package"
|
||||||
|
assert origin.name == "test-modules"
|
||||||
|
|
||||||
|
|
||||||
|
def test_distribution_commit_read_from_direct_url():
|
||||||
|
class FakeDistribution:
|
||||||
|
def read_text(self, filename):
|
||||||
|
assert filename == "direct_url.json"
|
||||||
|
return json.dumps(
|
||||||
|
{
|
||||||
|
"url": "https://github.com/example/example-modules",
|
||||||
|
"vcs_info": {"commit_id": "abc1234", "vcs": "git"},
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
assert module_loader._distribution_commit(FakeDistribution()) == "abc1234"
|
||||||
|
|
||||||
|
|
||||||
|
def test_command_log_records_loaded_modules(tmp_path):
|
||||||
|
(tmp_path / "Manifest.db").touch()
|
||||||
|
(tmp_path / "Info.plist").touch()
|
||||||
|
module_path = _write_custom_module(
|
||||||
|
tmp_path / "custom.py",
|
||||||
|
"AuditedRunModule",
|
||||||
|
(("ios", "check-backup"),),
|
||||||
|
slug="audited_run_module",
|
||||||
|
)
|
||||||
|
file_sha256 = hashlib.sha256(module_path.read_bytes()).hexdigest()
|
||||||
|
output_path = tmp_path / "out"
|
||||||
|
|
||||||
|
result = CliRunner().invoke(
|
||||||
|
check_backup,
|
||||||
|
[
|
||||||
|
"--module",
|
||||||
|
"AuditedRunModule",
|
||||||
|
"--load-module",
|
||||||
|
str(module_path),
|
||||||
|
"--output",
|
||||||
|
str(output_path),
|
||||||
|
str(tmp_path),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
|
||||||
|
assert result.exit_code == 0
|
||||||
|
command_log = (output_path / "command.log").read_text(encoding="utf-8")
|
||||||
|
assert (
|
||||||
|
f"Loaded 1 check-backup modules from '{module_path}' "
|
||||||
|
f"(sha256: {file_sha256}): AuditedRunModule" in command_log
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
class NestedBugreportModule(MVTModule):
|
class NestedBugreportModule(MVTModule):
|
||||||
supported_commands = (("android", "check-bugreport"),)
|
supported_commands = (("android", "check-bugreport"),)
|
||||||
|
|
||||||
|
|||||||
Reference in new issue
Block a user