mirror of
https://github.com/mvt-project/mvt.git
synced 2026-10-26 16:03:52 +01:00
Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f7e48f0c65 | ||
|
|
a66eeaa089 | ||
|
|
8685d1afcf |
No files matched your search
@@ -82,6 +82,11 @@ Module-running `check-*` commands can load custom Python modules with
|
|||||||
[development documentation](https://docs.mvt.re/en/latest/development/) for
|
[development documentation](https://docs.mvt.re/en/latest/development/) for
|
||||||
details.
|
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
|
## License
|
||||||
|
|
||||||
|
|||||||
@@ -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`.
|
||||||
@@ -31,6 +31,7 @@ nav:
|
|||||||
- Introduction: "introduction.md"
|
- Introduction: "introduction.md"
|
||||||
- Installation: "install.md"
|
- Installation: "install.md"
|
||||||
- Command Completion: "command_completion.md"
|
- Command Completion: "command_completion.md"
|
||||||
|
- Custom CLI Commands: "custom_commands.md"
|
||||||
- Using Docker: "docker.md"
|
- Using Docker: "docker.md"
|
||||||
- MVT for iOS:
|
- MVT for iOS:
|
||||||
- iOS Forensic Methodology: "ios/methodology.md"
|
- iOS Forensic Methodology: "ios/methodology.md"
|
||||||
|
|||||||
@@ -8,6 +8,12 @@ from zipfile import BadZipFile
|
|||||||
|
|
||||||
import click
|
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.cmd_check_iocs import CmdCheckIOCS
|
||||||
from mvt.common.completion import (
|
from mvt.common.completion import (
|
||||||
SUPPORTED_SHELLS,
|
SUPPORTED_SHELLS,
|
||||||
@@ -82,6 +88,7 @@ def _load_custom_modules(load_module):
|
|||||||
# Main
|
# Main
|
||||||
# ==============================================================================
|
# ==============================================================================
|
||||||
@click.group(invoke_without_command=False)
|
@click.group(invoke_without_command=False)
|
||||||
|
@load_cli_commands_option
|
||||||
@click.option(
|
@click.option(
|
||||||
"--disable-update-check", is_flag=True, help=HELP_MSG_DISABLE_UPDATE_CHECK
|
"--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():
|
def download_indicators():
|
||||||
ioc_updates = IndicatorsUpdates()
|
ioc_updates = IndicatorsUpdates()
|
||||||
ioc_updates.update()
|
ioc_updates.update()
|
||||||
|
|
||||||
|
|
||||||
|
register_cli_plugins(
|
||||||
|
cli,
|
||||||
|
entry_point_group=ANDROID_CLI_PLUGIN_GROUP,
|
||||||
|
environment_variable=MVT_ANDROID_CUSTOM_COMMANDS_ENV,
|
||||||
|
)
|
||||||
@@ -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)"
|
||||||
|
),
|
||||||
|
)
|
||||||
@@ -8,6 +8,12 @@ import logging
|
|||||||
import os
|
import os
|
||||||
|
|
||||||
import click
|
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.cmd_check_iocs import CmdCheckIOCS
|
||||||
from mvt.common.completion import (
|
from mvt.common.completion import (
|
||||||
SUPPORTED_SHELLS,
|
SUPPORTED_SHELLS,
|
||||||
@@ -86,6 +92,7 @@ def _load_custom_modules(load_module):
|
|||||||
# Main
|
# Main
|
||||||
# ==============================================================================
|
# ==============================================================================
|
||||||
@click.group(invoke_without_command=False)
|
@click.group(invoke_without_command=False)
|
||||||
|
@load_cli_commands_option
|
||||||
@click.option(
|
@click.option(
|
||||||
"--disable-update-check", is_flag=True, help=HELP_MSG_DISABLE_UPDATE_CHECK
|
"--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():
|
def download_iocs():
|
||||||
ioc_updates = IndicatorsUpdates()
|
ioc_updates = IndicatorsUpdates()
|
||||||
ioc_updates.update()
|
ioc_updates.update()
|
||||||
|
|
||||||
|
|
||||||
|
register_cli_plugins(
|
||||||
|
cli,
|
||||||
|
entry_point_group=IOS_CLI_PLUGIN_GROUP,
|
||||||
|
environment_variable=MVT_IOS_CUSTOM_COMMANDS_ENV,
|
||||||
|
)
|
||||||
@@ -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"}
|
||||||
Reference in new issue
Block a user