Add extensible CLI commands (#853)

* Add extensible CLI commands

* Handle plugin SystemExit failures
This commit is contained in:
besendorf
2026-08-05 23:30:28 +02:00
committed by GitHub
parent 93b7fb5232
commit 067f053627
7 changed files with 809 additions and 0 deletions
+5
View File
@@ -82,6 +82,11 @@ Module-running `check-*` commands can load custom Python modules with
[development documentation](https://docs.mvt.re/en/latest/development/) for
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/)
for the plugin entry points and `--load-command` interface.
## License
+107
View File
@@ -0,0 +1,107 @@
# 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):
commands add new CLI operations, while modules add analysis steps to existing
`check-*` commands.
!!! warning
Custom commands run as trusted Python code inside the MVT process. Install
or load commands only from sources you trust. MVT does not sandbox
third-party commands, and the MVT maintainers do not maintain them.
## Install a Command Package
Python packages can register a Click command or group for either MVT CLI. A
minimal package can expose this command from `my_mvt_plugin.py`:
```python
import click
@click.command()
@click.argument("path", type=click.Path(exists=True))
def summarize(path):
"""Summarize an acquisition."""
click.echo(f"Summarizing {path}")
```
Register the object in the package's `pyproject.toml`. The entry-point name is
the command users invoke:
```toml
[project.entry-points."mvt.ios.cli_plugins"]
summarize = "my_mvt_plugin:summarize"
[project.entry-points."mvt.android.cli_plugins"]
summarize = "my_mvt_plugin:summarize"
```
Use only the iOS or Android group if the command is platform-specific. After
installing the package in the same environment as MVT, it appears directly in
the appropriate CLI:
```bash
mvt-ios summarize ./ios-backup
mvt-android summarize ./androidqf-output
```
For a `pipx` installation of MVT, inject the plugin into MVT's environment:
```bash
pipx inject mvt my-mvt-plugin
```
When MVT is installed in an active virtual environment, install the plugin with
`pip` in that environment.
## Load a Command File
For local commands that are not packaged, create a Python file that exports one
Click command or group named `cli`:
```python
import click
@click.command("case-summary")
@click.argument("path", type=click.Path(exists=True))
def cli(path):
"""Summarize a case directory."""
click.echo(f"Summarizing {path}")
```
Pass the file before the custom command name:
```bash
mvt-ios --load-command ./case_summary.py case-summary ./ios-backup
```
`--load-command` can be repeated and also accepts a folder. MVT loads
non-hidden top-level `*.py` files in sorted order and skips `__init__.py`.
Every loaded file must export one `cli` object.
To load a file or folder on every invocation, set the platform-specific
environment variable:
```bash
export MVT_IOS_CUSTOM_COMMANDS=./ios_commands
export MVT_ANDROID_CUSTOM_COMMANDS=./android_commands
```
## Naming and Errors
Built-in MVT commands cannot be replaced. External command names must also be
unique; when installed packages or environment paths collide, MVT keeps the
first command and logs a warning. A collision from an explicit
`--load-command` is a usage error.
A package entry point or environment command that cannot be imported appears
as a marked broken command without preventing other MVT commands from working.
Invoke that command to see its package or file source and the underlying error.
An invalid command supplied explicitly with `--load-command` fails immediately
with a usage error.
Installed command packages use the entry-point name as the CLI command name.
The entry point must resolve to a `click.Command` or `click.Group`.
+1
View File
@@ -31,6 +31,7 @@ nav:
- Introduction: "introduction.md"
- Installation: "install.md"
- Command Completion: "command_completion.md"
- Custom CLI Commands: "custom_commands.md"
- Using Docker: "docker.md"
- MVT for iOS:
- iOS Forensic Methodology: "ios/methodology.md"
+14
View File
@@ -8,6 +8,12 @@ from zipfile import BadZipFile
import click
from mvt.common.cli_plugins import (
ANDROID_CLI_PLUGIN_GROUP,
MVT_ANDROID_CUSTOM_COMMANDS_ENV,
load_cli_commands_option,
register_cli_plugins,
)
from mvt.common.cmd_check_iocs import CmdCheckIOCS
from mvt.common.completion import (
SUPPORTED_SHELLS,
@@ -82,6 +88,7 @@ def _load_custom_modules(load_module):
# Main
# ==============================================================================
@click.group(invoke_without_command=False)
@load_cli_commands_option
@click.option(
"--disable-update-check", is_flag=True, help=HELP_MSG_DISABLE_UPDATE_CHECK
)
@@ -505,3 +512,10 @@ def check_iocs(ctx, iocs, list_modules, module, load_module, folder):
def download_indicators():
ioc_updates = IndicatorsUpdates()
ioc_updates.update()
register_cli_plugins(
cli,
entry_point_group=ANDROID_CLI_PLUGIN_GROUP,
environment_variable=MVT_ANDROID_CUSTOM_COMMANDS_ENV,
)
+297
View File
@@ -0,0 +1,297 @@
# Mobile Verification Toolkit (MVT)
# Copyright (c) 2021-2026 The MVT Authors.
# Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/
import hashlib
import importlib.metadata
import importlib.util
import logging
import os
import sys
from pathlib import Path
from types import ModuleType
from typing import Iterable
import click
IOS_CLI_PLUGIN_GROUP = "mvt.ios.cli_plugins"
ANDROID_CLI_PLUGIN_GROUP = "mvt.android.cli_plugins"
MVT_IOS_CUSTOM_COMMANDS_ENV = "MVT_IOS_CUSTOM_COMMANDS"
MVT_ANDROID_CUSTOM_COMMANDS_ENV = "MVT_ANDROID_CUSTOM_COMMANDS"
log = logging.getLogger(__name__)
class CustomCommandLoadError(Exception):
pass
class BrokenPluginCommand(click.Command):
"""A placeholder for an installed or configured command that failed to load."""
def __init__(self, name: str, source: str, exception: BaseException):
super().__init__(
name,
help=(
f"Unable to load external command from {source}.\n\n"
f"{type(exception).__name__}: {exception}"
),
short_help="Warning: external command could not be loaded.",
)
self.source = source
self.exception = exception
def parse_args(self, ctx: click.Context, args: list[str]) -> list[str]:
return args
def invoke(self, ctx: click.Context) -> None:
raise click.ClickException(
f"Unable to load external command '{self.name}' from {self.source}: "
f"{type(self.exception).__name__}: {self.exception}"
)
def _module_name_for_path(path: Path) -> str:
digest = hashlib.sha256(str(path).encode("utf-8")).hexdigest()[:16]
return f"_mvt_custom_command_{path.stem}_{digest}"
def _iter_command_files(path: Path) -> Iterable[Path]:
if path.is_file():
if path.suffix != ".py":
raise CustomCommandLoadError(
f"Custom command file is not a Python file: {path}"
)
yield path
return
if path.is_dir():
for child in sorted(path.iterdir()):
if child.name.startswith(".") or child.name == "__init__.py":
continue
if child.is_file() and child.suffix == ".py":
yield child
return
raise CustomCommandLoadError(f"Custom command path does not exist: {path}")
def _load_python_file(path: Path) -> ModuleType:
module_name = _module_name_for_path(path)
spec = importlib.util.spec_from_file_location(module_name, path)
if spec is None or spec.loader is None:
raise CustomCommandLoadError(f"Unable to load custom command file: {path}")
module = importlib.util.module_from_spec(spec)
sys.modules[module_name] = module
try:
spec.loader.exec_module(module)
except (Exception, SystemExit) as exc:
raise CustomCommandLoadError(
f"Unable to import custom command {path}: {exc}"
) from exc
return module
def load_cli_command_file(path: Path) -> click.Command:
module = _load_python_file(path)
command = getattr(module, "cli", None)
if not isinstance(command, click.Command):
raise CustomCommandLoadError(
f"Custom command {path} must export a Click command or group named 'cli'"
)
if command.name is None:
raise CustomCommandLoadError(f"Custom command {path} has no command name")
return command
def _register_command(
group: click.Group,
command: click.Command,
*,
name: str,
source: str,
collision_is_error: bool,
) -> bool:
if name in group.commands:
registered_sources = getattr(group, "_mvt_external_command_sources", {})
if registered_sources.get(name) == source:
return False
message = (
f"Unable to register external command '{name}' from {source}: "
"the command name is already registered"
)
if collision_is_error:
raise CustomCommandLoadError(message)
log.warning(message)
return False
group.add_command(command, name=name)
registered_sources = getattr(group, "_mvt_external_command_sources", {})
registered_sources[name] = source
setattr(group, "_mvt_external_command_sources", registered_sources)
return True
def register_cli_commands_from_path(
group: click.Group,
path: str | Path,
*,
collision_is_error: bool = False,
failures_are_errors: bool = False,
) -> list[str]:
resolved_path = Path(path).expanduser().resolve()
registered: list[str] = []
try:
command_files = list(_iter_command_files(resolved_path))
except CustomCommandLoadError as exc:
if failures_are_errors:
raise
name = resolved_path.stem.replace("_", "-") or "custom-command"
if _register_command(
group,
BrokenPluginCommand(name, str(resolved_path), exc),
name=name,
source=str(resolved_path),
collision_is_error=collision_is_error,
):
registered.append(name)
return registered
for command_file in command_files:
try:
command = load_cli_command_file(command_file)
except CustomCommandLoadError as exc:
if failures_are_errors:
raise
name = command_file.stem.replace("_", "-")
command = BrokenPluginCommand(name, str(command_file), exc)
command_name = command.name
if command_name is None:
raise CustomCommandLoadError(
f"Custom command {command_file} has no command name"
)
if _register_command(
group,
command,
name=command_name,
source=str(command_file),
collision_is_error=collision_is_error,
):
registered.append(command_name)
return registered
def _entry_point_source(entry_point: importlib.metadata.EntryPoint) -> str:
try:
distribution = getattr(entry_point, "dist", None)
if distribution is None:
return f"entry point {entry_point.value}"
name = distribution.metadata.get("Name", "unknown distribution")
version = getattr(distribution, "version", None)
if version:
return f"{name} {version} ({entry_point.value})"
return f"{name} ({entry_point.value})"
except Exception:
return f"entry point {entry_point.value}"
def register_installed_cli_commands(
group: click.Group,
entry_point_group: str,
) -> list[str]:
try:
entry_points = importlib.metadata.entry_points(group=entry_point_group)
except Exception as exc:
log.warning(
"Unable to discover external commands in entry-point group %s: %s",
entry_point_group,
exc,
)
return []
ordered_entry_points = sorted(
entry_points,
key=lambda entry_point: (
entry_point.name,
_entry_point_source(entry_point),
entry_point.value,
),
)
registered: list[str] = []
for entry_point in ordered_entry_points:
source = _entry_point_source(entry_point)
try:
command = entry_point.load()
if not isinstance(command, click.Command):
raise TypeError(
f"entry point must resolve to a Click command or group, "
f"not {type(command).__name__}"
)
except (Exception, SystemExit) as exc:
command = BrokenPluginCommand(entry_point.name, source, exc)
if _register_command(
group,
command,
name=entry_point.name,
source=source,
collision_is_error=False,
):
registered.append(entry_point.name)
return registered
def register_cli_plugins(
group: click.Group,
*,
entry_point_group: str,
environment_variable: str,
) -> None:
environment_path = os.environ.get(environment_variable)
if environment_path:
register_cli_commands_from_path(group, environment_path)
register_installed_cli_commands(group, entry_point_group)
def _load_command_option_callback(
ctx: click.Context,
param: click.Parameter,
paths: tuple[str, ...],
) -> tuple[str, ...]:
if not isinstance(ctx.command, click.Group):
raise click.ClickException("--load-command requires a Click command group")
for path in paths:
try:
register_cli_commands_from_path(
ctx.command,
path,
collision_is_error=True,
failures_are_errors=True,
)
except CustomCommandLoadError as exc:
raise click.BadParameter(str(exc), ctx=ctx, param=param) from exc
return paths
load_cli_commands_option = click.option(
"--load-command",
"load_commands",
type=click.Path(exists=True),
multiple=True,
expose_value=False,
is_eager=True,
callback=_load_command_option_callback,
help=(
"Load a custom CLI command from a Python file or folder "
"(can be invoked multiple times)"
),
)
+14
View File
@@ -8,6 +8,12 @@ import logging
import os
import click
from mvt.common.cli_plugins import (
IOS_CLI_PLUGIN_GROUP,
MVT_IOS_CUSTOM_COMMANDS_ENV,
load_cli_commands_option,
register_cli_plugins,
)
from mvt.common.cmd_check_iocs import CmdCheckIOCS
from mvt.common.completion import (
SUPPORTED_SHELLS,
@@ -86,6 +92,7 @@ def _load_custom_modules(load_module):
# Main
# ==============================================================================
@click.group(invoke_without_command=False)
@load_cli_commands_option
@click.option(
"--disable-update-check", is_flag=True, help=HELP_MSG_DISABLE_UPDATE_CHECK
)
@@ -523,3 +530,10 @@ def check_iocs(ctx, iocs, list_modules, module, load_module, folder):
def download_iocs():
ioc_updates = IndicatorsUpdates()
ioc_updates.update()
register_cli_plugins(
cli,
entry_point_group=IOS_CLI_PLUGIN_GROUP,
environment_variable=MVT_IOS_CUSTOM_COMMANDS_ENV,
)
+371
View File
@@ -0,0 +1,371 @@
from types import SimpleNamespace
import click
from click.testing import CliRunner
from mvt.common.cli_plugins import (
ANDROID_CLI_PLUGIN_GROUP,
IOS_CLI_PLUGIN_GROUP,
BrokenPluginCommand,
load_cli_commands_option,
register_cli_commands_from_path,
register_cli_plugins,
register_installed_cli_commands,
)
COMMAND_TEMPLATE = """
import click
@click.command({name!r})
@click.pass_context
def cli(ctx):
click.echo({message!r})
if ctx.obj:
click.echo(ctx.obj.get("marker", ""))
"""
def _write_command(path, name, message="command ran"):
path.write_text(
COMMAND_TEMPLATE.format(name=name, message=message),
encoding="utf-8",
)
return path
def _make_group():
@click.group()
@load_cli_commands_option
@click.pass_context
def group(ctx):
ctx.ensure_object(dict)
ctx.obj["marker"] = "parent context"
return group
def _entry_point(name, value, command=None, exception=None, distribution="plugin"):
def load():
if exception is not None:
raise exception
return command
dist = SimpleNamespace(metadata={"Name": distribution}, version="1.0")
return SimpleNamespace(name=name, value=value, load=load, dist=dist)
def test_load_command_option_registers_command_before_resolution(tmp_path):
command_path = _write_command(tmp_path / "hello.py", "hello")
group = _make_group()
result = CliRunner().invoke(
group,
["--load-command", str(command_path), "hello"],
)
assert result.exit_code == 0
assert "command ran" in result.output
assert "parent context" in result.output
def test_load_command_option_supports_folders_and_repeated_paths(tmp_path):
folder = tmp_path / "commands"
folder.mkdir()
_write_command(folder / "b.py", "second")
_write_command(folder / "a.py", "first")
_write_command(folder / ".hidden.py", "hidden")
_write_command(folder / "__init__.py", "init")
other = _write_command(tmp_path / "third.py", "third")
group = _make_group()
result = CliRunner().invoke(
group,
[
"--load-command",
str(folder),
"--load-command",
str(other),
"--load-command",
str(other),
"--help",
],
)
assert result.exit_code == 0
assert "first" in result.output
assert "second" in result.output
assert "third" in result.output
assert "hidden" not in result.output
assert "init" not in result.output
def test_loaded_command_participates_in_shell_completion(tmp_path):
command_path = _write_command(tmp_path / "hello.py", "hello")
group = _make_group()
words = f"group --load-command {command_path} he"
result = CliRunner().invoke(
group,
[],
env={
"_GROUP_COMPLETE": "bash_complete",
"COMP_WORDS": words,
"COMP_CWORD": "3",
},
)
assert result.exit_code == 0
assert "plain,hello" in result.output
def test_explicit_command_import_and_contract_failures_are_usage_errors(tmp_path):
broken_path = tmp_path / "broken.py"
broken_path.write_text("raise RuntimeError('broken import')", encoding="utf-8")
missing_cli_path = tmp_path / "missing_cli.py"
missing_cli_path.write_text("value = 1", encoding="utf-8")
broken_result = CliRunner().invoke(
_make_group(),
["--load-command", str(broken_path), "broken"],
)
missing_cli_result = CliRunner().invoke(
_make_group(),
["--load-command", str(missing_cli_path), "missing-cli"],
)
assert broken_result.exit_code == 2
assert "broken import" in broken_result.output
assert missing_cli_result.exit_code == 2
assert "must export a Click command or group named 'cli'" in (
missing_cli_result.output
)
def test_environment_command_failure_gets_broken_placeholder(tmp_path):
command_path = tmp_path / "broken_command.py"
command_path.write_text("raise RuntimeError('broken import')", encoding="utf-8")
group = click.Group()
registered = register_cli_commands_from_path(group, command_path)
assert registered == ["broken-command"]
assert isinstance(group.commands["broken-command"], BrokenPluginCommand)
result = CliRunner().invoke(group, ["broken-command"])
assert result.exit_code == 1
assert "broken import" in result.output
assert str(command_path) in result.output
def test_environment_command_system_exit_gets_broken_placeholder(tmp_path):
command_path = tmp_path / "exiting_command.py"
command_path.write_text("raise SystemExit(7)", encoding="utf-8")
group = click.Group()
registered = register_cli_commands_from_path(group, command_path)
assert registered == ["exiting-command"]
assert isinstance(group.commands["exiting-command"], BrokenPluginCommand)
result = CliRunner().invoke(group, ["exiting-command"])
assert result.exit_code == 1
assert "Unable to import custom command" in result.output
assert result.output.rstrip().endswith(": 7")
def test_installed_entry_point_name_is_the_command_name(monkeypatch):
@click.command("internal-name")
def command():
click.echo("installed command ran")
entry_point = _entry_point(
"external-name",
"example_plugin:cli",
command=command,
)
monkeypatch.setattr(
"mvt.common.cli_plugins.importlib.metadata.entry_points",
lambda **kwargs: [entry_point],
)
group = click.Group()
registered = register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
assert registered == ["external-name"]
assert "internal-name" not in group.commands
result = CliRunner().invoke(group, ["external-name"])
assert result.exit_code == 0
assert result.output == "installed command ran\n"
def test_broken_installed_plugin_does_not_break_cli(monkeypatch):
broken = _entry_point(
"broken",
"broken_plugin:cli",
exception=RuntimeError("missing dependency"),
distribution="broken-plugin",
)
monkeypatch.setattr(
"mvt.common.cli_plugins.importlib.metadata.entry_points",
lambda **kwargs: [broken],
)
group = click.Group()
register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
help_result = CliRunner().invoke(group, ["--help"])
assert help_result.exit_code == 0
assert "Warning: external command could not be loaded." in help_result.output
result = CliRunner().invoke(group, ["broken"])
assert result.exit_code == 1
assert "broken-plugin 1.0 (broken_plugin:cli)" in result.output
assert "RuntimeError: missing dependency" in result.output
def test_installed_plugin_system_exit_does_not_break_cli(monkeypatch):
exiting = _entry_point(
"exiting",
"exiting_plugin:cli",
exception=SystemExit(7),
distribution="exiting-plugin",
)
monkeypatch.setattr(
"mvt.common.cli_plugins.importlib.metadata.entry_points",
lambda **kwargs: [exiting],
)
group = click.Group()
register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
help_result = CliRunner().invoke(group, ["--help"])
assert help_result.exit_code == 0
result = CliRunner().invoke(group, ["exiting"])
assert result.exit_code == 1
assert "SystemExit: 7" in result.output
def test_non_click_entry_point_gets_broken_placeholder(monkeypatch):
invalid = _entry_point("invalid", "plugin:value", command=object())
monkeypatch.setattr(
"mvt.common.cli_plugins.importlib.metadata.entry_points",
lambda **kwargs: [invalid],
)
group = click.Group()
register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
assert isinstance(group.commands["invalid"], BrokenPluginCommand)
result = CliRunner().invoke(group, ["invalid"])
assert result.exit_code == 1
assert "must resolve to a Click command or group" in result.output
def test_entry_point_discovery_failure_does_not_break_group(monkeypatch, caplog):
def fail_discovery(**kwargs):
raise RuntimeError("invalid package metadata")
monkeypatch.setattr(
"mvt.common.cli_plugins.importlib.metadata.entry_points",
fail_discovery,
)
group = click.Group()
registered = register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
assert registered == []
assert not group.commands
assert "Unable to discover external commands" in caplog.text
assert "invalid package metadata" in caplog.text
def test_builtin_and_first_external_command_win_collisions(monkeypatch, caplog):
@click.command("version")
def core_version():
pass
@click.command()
def first():
pass
@click.command()
def second():
pass
entry_points = [
_entry_point("duplicate", "z_plugin:cli", command=second, distribution="z"),
_entry_point("version", "plugin:version", command=first),
_entry_point("duplicate", "a_plugin:cli", command=first, distribution="a"),
]
monkeypatch.setattr(
"mvt.common.cli_plugins.importlib.metadata.entry_points",
lambda **kwargs: entry_points,
)
group = click.Group(commands={"version": core_version})
registered = register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
assert registered == ["duplicate"]
assert group.commands["version"] is core_version
assert group.commands["duplicate"] is first
assert "the command name is already registered" in caplog.text
def test_explicit_command_cannot_replace_existing_command(tmp_path):
command_path = _write_command(tmp_path / "version.py", "version")
group = _make_group()
@group.command("version")
def core_version():
pass
result = CliRunner().invoke(
group,
["--load-command", str(command_path), "version"],
)
assert result.exit_code == 2
assert "the command name is already registered" in result.output
def test_platform_entry_point_groups_and_environment_paths_are_separate(
tmp_path, monkeypatch
):
ios_path = _write_command(tmp_path / "ios.py", "ios-file")
android_path = _write_command(tmp_path / "android.py", "android-file")
@click.command()
def ios_package():
pass
@click.command()
def android_package():
pass
def entry_points(*, group):
if group == IOS_CLI_PLUGIN_GROUP:
return [_entry_point("ios-package", "ios_plugin:cli", ios_package)]
return [_entry_point("android-package", "android_plugin:cli", android_package)]
monkeypatch.setattr(
"mvt.common.cli_plugins.importlib.metadata.entry_points",
entry_points,
)
monkeypatch.setenv("TEST_IOS_COMMANDS", str(ios_path))
monkeypatch.setenv("TEST_ANDROID_COMMANDS", str(android_path))
ios_group = click.Group()
android_group = click.Group()
register_cli_plugins(
ios_group,
entry_point_group=IOS_CLI_PLUGIN_GROUP,
environment_variable="TEST_IOS_COMMANDS",
)
register_cli_plugins(
android_group,
entry_point_group=ANDROID_CLI_PLUGIN_GROUP,
environment_variable="TEST_ANDROID_COMMANDS",
)
assert set(ios_group.commands) == {"ios-file", "ios-package"}
assert set(android_group.commands) == {"android-file", "android-package"}