mirror of
https://github.com/mvt-project/mvt.git
synced 2026-10-22 15:18:26 +02:00
Compare commits
6
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f8456059db | ||
|
|
efbab29f94 | ||
|
|
ecc22f54b7 | ||
|
|
90fc0ae351 | ||
|
|
dab82e38d0 | ||
|
|
4e031cd43f |
No files matched your search
@@ -84,7 +84,7 @@ details.
|
||||
|
||||
Users can also add top-level commands to `mvt-ios` and `mvt-android` from
|
||||
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.
|
||||
|
||||
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
# Custom CLI Commands
|
||||
|
||||
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
|
||||
`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
|
||||
`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
|
||||
|
||||
For local commands that are not packaged, create a Python file that exports one
|
||||
@@ -35,9 +35,16 @@ class DependentModule(MVTModule):
|
||||
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.
|
||||
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
|
||||
|
||||
@@ -160,6 +167,10 @@ For a `pipx` installation of MVT, inject the package into MVT's environment:
|
||||
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>`),
|
||||
@@ -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"
|
||||
- Indicators of Compromise: "iocs.md"
|
||||
- Development:
|
||||
- Development Instructions: "development.md"
|
||||
- Custom CLI Commands: "custom_commands.md"
|
||||
- Development Instructions: "development/index.md"
|
||||
- Custom CLI Commands: "development/custom_commands.md"
|
||||
- Plugin Configuration: "development/plugin_configuration.md"
|
||||
- License: "license.md"
|
||||
+2
-2
@@ -44,8 +44,8 @@ homepage = "https://docs.mvt.re/en/latest/"
|
||||
repository = "https://github.com/mvt-project/mvt"
|
||||
|
||||
[project.scripts]
|
||||
mvt-ios = "mvt.ios:cli"
|
||||
mvt-android = "mvt.android:cli"
|
||||
mvt-ios = "mvt.ios:main"
|
||||
mvt-android = "mvt.android:main"
|
||||
|
||||
[dependency-groups]
|
||||
dev = [
|
||||
|
||||
@@ -3,4 +3,4 @@
|
||||
# Use of this software is governed by the MVT License 1.1 that can be found at
|
||||
# https://license.mvt.re/1.1/
|
||||
|
||||
from .cli import cli
|
||||
from .cli import cli, main
|
||||
+16
-5
@@ -514,8 +514,19 @@ def download_indicators():
|
||||
ioc_updates.update()
|
||||
|
||||
|
||||
register_cli_plugins(
|
||||
cli,
|
||||
entry_point_group=ANDROID_CLI_PLUGIN_GROUP,
|
||||
environment_variable=MVT_ANDROID_CUSTOM_COMMANDS_ENV,
|
||||
)
|
||||
# ==============================================================================
|
||||
# Entry point of the mvt-android console script
|
||||
# ==============================================================================
|
||||
def main() -> None:
|
||||
"""Register the external commands and run the mvt-android CLI.
|
||||
|
||||
External commands are registered here rather than when this module is
|
||||
imported, so that importing MVT never runs third-party code and a plugin
|
||||
importing from MVT cannot re-enter a module that is still initializing.
|
||||
"""
|
||||
register_cli_plugins(
|
||||
cli,
|
||||
entry_point_group=ANDROID_CLI_PLUGIN_GROUP,
|
||||
environment_variable=MVT_ANDROID_CUSTOM_COMMANDS_ENV,
|
||||
)
|
||||
cli()
|
||||
+95
-16
@@ -317,6 +317,81 @@ class Command:
|
||||
console.print("")
|
||||
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]]]:
|
||||
"""Return enabled modules in stable topological order."""
|
||||
modules = self._available_modules()
|
||||
@@ -329,30 +404,34 @@ class Command:
|
||||
else:
|
||||
selected = [module for module in modules if module.enabled]
|
||||
|
||||
required = set(selected)
|
||||
required: set[type[MVTModule]] = set()
|
||||
pending = list(selected)
|
||||
while pending:
|
||||
module = pending.pop()
|
||||
if module in required:
|
||||
continue
|
||||
required.add(module)
|
||||
for dependency in module.dependencies:
|
||||
if dependency not in module_indexes:
|
||||
self.log.warning(
|
||||
"Module %s depends on unavailable module %s. "
|
||||
"No modules will be run.",
|
||||
module.__name__,
|
||||
dependency.__name__,
|
||||
)
|
||||
return None
|
||||
if dependency not in required:
|
||||
required.add(dependency)
|
||||
# Unavailable dependencies are reported by _skipped_modules().
|
||||
if dependency in module_indexes:
|
||||
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]]] = {
|
||||
module: [] for module in required
|
||||
module: [] for module in runnable
|
||||
}
|
||||
indegree = {module: 0 for module in required}
|
||||
for module in required:
|
||||
indegree = {module: 0 for module in runnable}
|
||||
for module in runnable:
|
||||
for dependency in module.dependencies:
|
||||
if dependency not in required:
|
||||
if dependency not in indegree:
|
||||
continue
|
||||
dependents[dependency].append(module)
|
||||
indegree[module] += 1
|
||||
@@ -371,7 +450,7 @@ class Command:
|
||||
if indegree[dependent] == 0:
|
||||
heappush(ready, (module_indexes[dependent], dependent))
|
||||
|
||||
if len(ordered) != len(required):
|
||||
if len(ordered) != len(runnable):
|
||||
cyclic_modules = sorted(
|
||||
(module.__name__ for module, count in indegree.items() if count > 0)
|
||||
)
|
||||
|
||||
@@ -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
|
||||
@@ -3,4 +3,4 @@
|
||||
# Use of this software is governed by the MVT License 1.1 that can be found at
|
||||
# https://license.mvt.re/1.1/
|
||||
|
||||
from .cli import cli
|
||||
from .cli import cli, main
|
||||
+16
-5
@@ -532,8 +532,19 @@ def download_iocs():
|
||||
ioc_updates.update()
|
||||
|
||||
|
||||
register_cli_plugins(
|
||||
cli,
|
||||
entry_point_group=IOS_CLI_PLUGIN_GROUP,
|
||||
environment_variable=MVT_IOS_CUSTOM_COMMANDS_ENV,
|
||||
)
|
||||
# ==============================================================================
|
||||
# Entry point of the mvt-ios console script
|
||||
# ==============================================================================
|
||||
def main() -> None:
|
||||
"""Register the external commands and run the mvt-ios CLI.
|
||||
|
||||
External commands are registered here rather than when this module is
|
||||
imported, so that importing MVT never runs third-party code and a plugin
|
||||
importing from MVT cannot re-enter a module that is still initializing.
|
||||
"""
|
||||
register_cli_plugins(
|
||||
cli,
|
||||
entry_point_group=IOS_CLI_PLUGIN_GROUP,
|
||||
environment_variable=MVT_IOS_CUSTOM_COMMANDS_ENV,
|
||||
)
|
||||
cli()
|
||||
@@ -157,7 +157,7 @@ class TestCommand:
|
||||
assert not hasattr(cmd, "initialized")
|
||||
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):
|
||||
pass
|
||||
|
||||
@@ -165,14 +165,97 @@ class TestCommand:
|
||||
dependencies = (UnavailableModule,)
|
||||
|
||||
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):
|
||||
cmd.run()
|
||||
|
||||
assert RecordingModule.run_order == []
|
||||
assert not hasattr(cmd, "initialized")
|
||||
assert "depends on unavailable module UnavailableModule" in caplog.text
|
||||
assert "Module DependentModule will be SKIPPED" 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):
|
||||
cmd = RecordingCommand()
|
||||
|
||||
@@ -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()
|
||||
@@ -0,0 +1,85 @@
|
||||
# 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/
|
||||
|
||||
"""Helpers building throwaway plugin distributions for the tests.
|
||||
|
||||
Some plugin behaviour only shows up in a fresh interpreter: what an import
|
||||
executes, and what a plugin sees when MVT is imported before or after it.
|
||||
These helpers write an importable distribution with a real entry point and
|
||||
run a script against it in a subprocess, with a temporary home so that the
|
||||
subprocess cannot touch the configuration of whoever runs the tests.
|
||||
"""
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
FIXTURE_COMMAND_NAME = "fixture-plugin"
|
||||
FIXTURE_MODULE_NAME = "fixture_cli_plugin"
|
||||
FIXTURE_DISTRIBUTION_NAME = "fixture-cli-plugin"
|
||||
|
||||
|
||||
def write_cli_plugin_distribution(
|
||||
site_path: Path,
|
||||
entry_point_group: str,
|
||||
module_source: str,
|
||||
) -> Path:
|
||||
"""Write a distribution registering a CLI plugin entry point.
|
||||
|
||||
:param site_path: Folder to write the distribution into, to be added to
|
||||
the import path of the interpreter loading it.
|
||||
:param entry_point_group: Entry-point group to register the command in.
|
||||
:param module_source: Source of the plugin module, which must define a
|
||||
Click command named `cli`.
|
||||
:returns: The folder the distribution was written to.
|
||||
"""
|
||||
site_path.mkdir(parents=True, exist_ok=True)
|
||||
(site_path / f"{FIXTURE_MODULE_NAME}.py").write_text(
|
||||
module_source, encoding="utf-8"
|
||||
)
|
||||
|
||||
dist_info = (
|
||||
site_path / f"{FIXTURE_DISTRIBUTION_NAME.replace('-', '_')}-1.0.dist-info"
|
||||
)
|
||||
dist_info.mkdir(exist_ok=True)
|
||||
(dist_info / "METADATA").write_text(
|
||||
f"Metadata-Version: 2.1\nName: {FIXTURE_DISTRIBUTION_NAME}\nVersion: 1.0\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
(dist_info / "entry_points.txt").write_text(
|
||||
f"[{entry_point_group}]\n{FIXTURE_COMMAND_NAME} = {FIXTURE_MODULE_NAME}:cli\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
return site_path
|
||||
|
||||
|
||||
def run_isolated_python(
|
||||
script: str,
|
||||
home: Path,
|
||||
site_path: Optional[Path] = None,
|
||||
**environment: str,
|
||||
) -> subprocess.CompletedProcess:
|
||||
"""Run a script in a fresh interpreter with its own configuration folder.
|
||||
|
||||
Importing MVT writes its configuration file, so the subprocess gets a
|
||||
temporary home and no MVT environment variables from the test session.
|
||||
"""
|
||||
isolated_environment = {
|
||||
key: value for key, value in os.environ.items() if not key.startswith("MVT_")
|
||||
}
|
||||
isolated_environment["HOME"] = str(home)
|
||||
isolated_environment["XDG_CONFIG_HOME"] = str(home / "config")
|
||||
if site_path is not None:
|
||||
isolated_environment["PYTHONPATH"] = str(site_path)
|
||||
isolated_environment.update(environment)
|
||||
|
||||
return subprocess.run(
|
||||
[sys.executable, "-c", script],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=isolated_environment,
|
||||
)
|
||||
@@ -0,0 +1,215 @@
|
||||
# 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 sys
|
||||
from types import SimpleNamespace
|
||||
|
||||
import click
|
||||
import pytest
|
||||
|
||||
import mvt.android
|
||||
import mvt.ios
|
||||
from mvt.android.cli import cli as android_cli
|
||||
from mvt.android.cli import main as android_main
|
||||
from mvt.common.cli_plugins import ANDROID_CLI_PLUGIN_GROUP, IOS_CLI_PLUGIN_GROUP
|
||||
from mvt.ios.cli import cli as ios_cli
|
||||
from mvt.ios.cli import main as ios_main
|
||||
|
||||
from .plugin_fixtures import (
|
||||
FIXTURE_COMMAND_NAME,
|
||||
run_isolated_python,
|
||||
write_cli_plugin_distribution,
|
||||
)
|
||||
|
||||
MARKER_PLUGIN_TEMPLATE = """
|
||||
import os
|
||||
|
||||
import click
|
||||
|
||||
# Touched when this module is imported, so a test can tell whether loading MVT
|
||||
# executed the plugin.
|
||||
open(os.environ["FIXTURE_PLUGIN_MARKER"], "a").close()
|
||||
|
||||
|
||||
@click.command()
|
||||
def cli():
|
||||
click.echo("fixture plugin ran")
|
||||
"""
|
||||
|
||||
PROGRAMS = {
|
||||
"mvt-ios": (mvt.ios, ios_cli, IOS_CLI_PLUGIN_GROUP),
|
||||
"mvt-android": (mvt.android, android_cli, ANDROID_CLI_PLUGIN_GROUP),
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def restore_cli_commands():
|
||||
"""Undo the plugin registration main() performs on the shared CLI groups."""
|
||||
originals = {
|
||||
program: dict(group.commands) for program, (_, group, _) in PROGRAMS.items()
|
||||
}
|
||||
yield
|
||||
for program, (_, group, _) in PROGRAMS.items():
|
||||
group.commands.clear()
|
||||
group.commands.update(originals[program])
|
||||
if hasattr(group, "_mvt_external_command_sources"):
|
||||
delattr(group, "_mvt_external_command_sources")
|
||||
|
||||
|
||||
def _install_fixture_entry_point(monkeypatch, entry_point_group, command):
|
||||
def entry_points(*, group):
|
||||
if group != entry_point_group:
|
||||
return []
|
||||
return [
|
||||
SimpleNamespace(
|
||||
name=FIXTURE_COMMAND_NAME,
|
||||
value="fixture_cli_plugin:cli",
|
||||
load=lambda: command,
|
||||
dist=SimpleNamespace(
|
||||
metadata={"Name": "fixture-cli-plugin"}, version="1.0"
|
||||
),
|
||||
)
|
||||
]
|
||||
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.cli_plugins.importlib.metadata.entry_points", entry_points
|
||||
)
|
||||
|
||||
|
||||
def _offline_argv(program, *arguments):
|
||||
"""Build an argument list which keeps the CLI from checking for updates."""
|
||||
return [
|
||||
program,
|
||||
"--disable-update-check",
|
||||
"--disable-indicator-update-check",
|
||||
*arguments,
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("program", sorted(PROGRAMS))
|
||||
def test_main_registers_installed_plugins_before_running_the_cli(
|
||||
program, monkeypatch, capsys, restore_cli_commands
|
||||
):
|
||||
package, group, entry_point_group = PROGRAMS[program]
|
||||
|
||||
@click.command()
|
||||
def fixture_command():
|
||||
click.echo("fixture plugin ran")
|
||||
|
||||
_install_fixture_entry_point(monkeypatch, entry_point_group, fixture_command)
|
||||
monkeypatch.setattr(sys, "argv", _offline_argv(program, FIXTURE_COMMAND_NAME))
|
||||
|
||||
with pytest.raises(SystemExit) as exit_info:
|
||||
package.main()
|
||||
|
||||
assert exit_info.value.code == 0
|
||||
assert "fixture plugin ran" in capsys.readouterr().out
|
||||
assert FIXTURE_COMMAND_NAME in group.commands
|
||||
|
||||
|
||||
@pytest.mark.parametrize("program", sorted(PROGRAMS))
|
||||
def test_main_completes_plugin_command_names(
|
||||
program, monkeypatch, capsys, restore_cli_commands
|
||||
):
|
||||
package, _, entry_point_group = PROGRAMS[program]
|
||||
|
||||
@click.command()
|
||||
def fixture_command():
|
||||
pass
|
||||
|
||||
_install_fixture_entry_point(monkeypatch, entry_point_group, fixture_command)
|
||||
complete_variable = f"_{program.upper().replace('-', '_')}_COMPLETE"
|
||||
monkeypatch.setenv(complete_variable, "bash_complete")
|
||||
monkeypatch.setenv("COMP_WORDS", f"{program} fixture")
|
||||
monkeypatch.setenv("COMP_CWORD", "1")
|
||||
monkeypatch.setattr(sys, "argv", [program])
|
||||
|
||||
with pytest.raises(SystemExit):
|
||||
package.main()
|
||||
|
||||
assert f"plain,{FIXTURE_COMMAND_NAME}" in capsys.readouterr().out
|
||||
|
||||
|
||||
@pytest.mark.parametrize("program", sorted(PROGRAMS))
|
||||
def test_main_still_loads_commands_from_a_file(
|
||||
program, monkeypatch, capsys, tmp_path, restore_cli_commands
|
||||
):
|
||||
package, _, entry_point_group = PROGRAMS[program]
|
||||
command_path = tmp_path / "case_summary.py"
|
||||
command_path.write_text(
|
||||
"import click\n"
|
||||
"\n"
|
||||
"\n"
|
||||
'@click.command("case-summary")\n'
|
||||
"def cli():\n"
|
||||
' click.echo("case summary ran")\n',
|
||||
encoding="utf-8",
|
||||
)
|
||||
_install_fixture_entry_point(
|
||||
monkeypatch, entry_point_group, click.Command("unused")
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
sys,
|
||||
"argv",
|
||||
_offline_argv(program, "--load-command", str(command_path), "case-summary"),
|
||||
)
|
||||
|
||||
with pytest.raises(SystemExit) as exit_info:
|
||||
package.main()
|
||||
|
||||
assert exit_info.value.code == 0
|
||||
assert "case summary ran" in capsys.readouterr().out
|
||||
|
||||
|
||||
def test_the_console_script_targets_are_importable():
|
||||
# [project.scripts] points at these, so they must stay on the packages.
|
||||
assert mvt.ios.main is ios_main
|
||||
assert mvt.android.main is android_main
|
||||
|
||||
|
||||
def test_importing_mvt_does_not_run_installed_plugins(tmp_path):
|
||||
site_path = write_cli_plugin_distribution(
|
||||
tmp_path / "site", IOS_CLI_PLUGIN_GROUP, MARKER_PLUGIN_TEMPLATE
|
||||
)
|
||||
marker = tmp_path / "plugin-imported"
|
||||
|
||||
result = run_isolated_python(
|
||||
"import mvt.ios.cli\nimport mvt.android.cli\nprint('imported')",
|
||||
home=tmp_path / "home",
|
||||
site_path=site_path,
|
||||
FIXTURE_PLUGIN_MARKER=str(marker),
|
||||
)
|
||||
|
||||
assert result.returncode == 0, result.stderr
|
||||
assert "imported" in result.stdout
|
||||
assert not marker.exists()
|
||||
|
||||
|
||||
def test_registering_the_plugins_runs_the_entry_point(tmp_path):
|
||||
site_path = write_cli_plugin_distribution(
|
||||
tmp_path / "site", IOS_CLI_PLUGIN_GROUP, MARKER_PLUGIN_TEMPLATE
|
||||
)
|
||||
marker = tmp_path / "plugin-imported"
|
||||
|
||||
result = run_isolated_python(
|
||||
"import click\n"
|
||||
"from mvt.common.cli_plugins import (\n"
|
||||
" IOS_CLI_PLUGIN_GROUP,\n"
|
||||
" BrokenPluginCommand,\n"
|
||||
" register_installed_cli_commands,\n"
|
||||
")\n"
|
||||
"group = click.Group()\n"
|
||||
"register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)\n"
|
||||
f"command = group.commands[{FIXTURE_COMMAND_NAME!r}]\n"
|
||||
"assert not isinstance(command, BrokenPluginCommand), command.help\n"
|
||||
"print('registered')\n",
|
||||
home=tmp_path / "home",
|
||||
site_path=site_path,
|
||||
FIXTURE_PLUGIN_MARKER=str(marker),
|
||||
)
|
||||
|
||||
assert result.returncode == 0, result.stderr
|
||||
assert "registered" in result.stdout
|
||||
assert marker.exists()
|
||||
Reference in new issue
Block a user