mirror of
https://github.com/mvt-project/mvt.git
synced 2026-10-23 07:38:22 +02:00
Compare commits
6
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
348b47db35 | ||
|
|
b768d5b86c | ||
|
|
c4215ec163 | ||
|
|
36d1b27c4c | ||
|
|
6070fc06fb | ||
|
|
c95e6659c6 |
No files matched your search
@@ -58,34 +58,31 @@ For alternative installation options and known issues, please refer to the [docu
|
||||
|
||||
## Usage
|
||||
|
||||
MVT provides two commands `mvt-ios` and `mvt-android`. [Check out the documentation to learn how to use them!](https://docs.mvt.re/)
|
||||
MVT provides three commands: `mvt-ios` and `mvt-android` analyse acquisitions from devices of that platform, and `mvt` hosts what belongs to neither: `version`, `completion`, `plugins` and `download-iocs` (`version` and `download-iocs` remain available on the platform commands for now). Running `mvt` on its own shows the installed version, update notices and the available commands. [Check out the documentation to learn how to use them!](https://docs.mvt.re/)
|
||||
|
||||
### Shell completion
|
||||
|
||||
MVT can generate shell completion scripts for Bash, Zsh, and Fish:
|
||||
MVT can generate a shell completion script for Bash, Zsh, and Fish which covers `mvt`, `mvt-ios` and `mvt-android`:
|
||||
|
||||
```bash
|
||||
mvt-ios completion
|
||||
mvt-android completion
|
||||
mvt completion
|
||||
```
|
||||
|
||||
The commands print setup instructions by default. To generate a completion script directly, pass the shell name:
|
||||
The command prints setup instructions by default. To generate the completion script directly, pass the shell name:
|
||||
|
||||
```bash
|
||||
mvt-ios completion bash
|
||||
mvt-android completion zsh
|
||||
mvt completion bash
|
||||
```
|
||||
|
||||
MVT only writes completion files or shell configuration when `--install` is passed. See the [command completion documentation](https://docs.mvt.re/en/latest/command_completion/) for details.
|
||||
Module-running `check-*` commands can load custom Python modules with
|
||||
`--load-module PATH` or from a folder set in `MVT_CUSTOM_MODULES`. See the
|
||||
[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
|
||||
Plugin packages extend MVT with additional forensic modules, which run inside
|
||||
the `check-*` commands, and with top-level commands on `mvt`, `mvt-ios` and
|
||||
`mvt-android`. See the
|
||||
[development documentation](https://docs.mvt.re/en/latest/development/) for
|
||||
writing and installing them, and the
|
||||
[custom CLI command documentation](https://docs.mvt.re/en/latest/development/custom_commands/)
|
||||
for the plugin entry points and `--load-command` interface.
|
||||
for the entry points a package registers commands in.
|
||||
|
||||
|
||||
## License
|
||||
|
||||
+22
-22
@@ -6,61 +6,61 @@ Click provides tab completion support for Bash (version 4.4 and up), Zsh, and Fi
|
||||
|
||||
To enable it, you need to register a completion script with your shell, which varies depending on the shell you are using.
|
||||
|
||||
The following describes how to generate the command completion scripts and add them to your shell configuration.
|
||||
`mvt completion` generates one script which covers `mvt`, `mvt-ios` and `mvt-android`. The following describes how to generate that script and add it to your shell configuration.
|
||||
|
||||
> **Note: You will need to start a new shell for the changes to take effect.**
|
||||
|
||||
### For Bash
|
||||
|
||||
```bash
|
||||
# Generate bash completion scripts
|
||||
mvt-ios completion bash > ~/.mvt-ios-complete.bash
|
||||
mvt-android completion bash > ~/.mvt-android-complete.bash
|
||||
# Generate the bash completion script
|
||||
mvt completion bash > ~/.mvt-complete.bash
|
||||
```
|
||||
|
||||
Add the following to `~/.bashrc`:
|
||||
```bash
|
||||
# source mvt completion scripts
|
||||
[ -f ~/.mvt-ios-complete.bash ] && . ~/.mvt-ios-complete.bash
|
||||
[ -f ~/.mvt-android-complete.bash ] && . ~/.mvt-android-complete.bash
|
||||
# source the mvt completion script
|
||||
[ -f ~/.mvt-complete.bash ] && . ~/.mvt-complete.bash
|
||||
```
|
||||
|
||||
### For Zsh
|
||||
|
||||
```bash
|
||||
# Generate zsh completion scripts
|
||||
mvt-ios completion zsh > ~/.mvt-ios-complete.zsh
|
||||
mvt-android completion zsh > ~/.mvt-android-complete.zsh
|
||||
# Generate the zsh completion script
|
||||
mvt completion zsh > ~/.mvt-complete.zsh
|
||||
```
|
||||
|
||||
Add the following to `~/.zshrc`:
|
||||
```bash
|
||||
# source mvt completion scripts
|
||||
[ -f ~/.mvt-ios-complete.zsh ] && . ~/.mvt-ios-complete.zsh
|
||||
[ -f ~/.mvt-android-complete.zsh ] && . ~/.mvt-android-complete.zsh
|
||||
# source the mvt completion script
|
||||
[ -f ~/.mvt-complete.zsh ] && . ~/.mvt-complete.zsh
|
||||
```
|
||||
|
||||
### For Fish
|
||||
|
||||
```bash
|
||||
# Generate fish completion scripts
|
||||
mkdir -p ~/.config/fish/completions
|
||||
mvt-ios completion fish > ~/.config/fish/completions/mvt-ios.fish
|
||||
mvt-android completion fish > ~/.config/fish/completions/mvt-android.fish
|
||||
# Generate the fish completion script
|
||||
mkdir -p ~/.config/fish/conf.d
|
||||
mvt completion fish > ~/.config/fish/conf.d/mvt-completion.fish
|
||||
```
|
||||
|
||||
Fish loads completion files from `~/.config/fish/completions` automatically.
|
||||
Fish loads the files in `~/.config/fish/conf.d` automatically.
|
||||
|
||||
### Automatic Installation
|
||||
|
||||
MVT can write the completion file and update the relevant shell configuration for Bash and Zsh when you pass `--install`:
|
||||
|
||||
```bash
|
||||
mvt-ios completion bash --install
|
||||
mvt-android completion bash --install
|
||||
mvt completion bash --install
|
||||
```
|
||||
|
||||
Replace `bash` with `zsh` or `fish` as needed. For Fish, `--install` writes the completion file into `~/.config/fish/completions`.
|
||||
Replace `bash` with `zsh` or `fish` as needed. For Fish, `--install` writes the completion file into `~/.config/fish/conf.d` and changes no shell configuration.
|
||||
|
||||
!!! note
|
||||
|
||||
Earlier versions generated one script per command, with `mvt-ios completion`
|
||||
and `mvt-android completion`. Files written by them keep working. When you
|
||||
switch to the single script, remove the old files and the lines which load
|
||||
them from your shell configuration.
|
||||
|
||||
For more information, visit the official [Click Docs](https://click.palletsprojects.com/en/stable/shell-completion/#enabling-completion).
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
# Custom CLI Commands
|
||||
|
||||
MVT can load additional top-level commands into `mvt-ios` and `mvt-android`.
|
||||
MVT can load additional top-level commands into `mvt`, `mvt-ios` and
|
||||
`mvt-android`. A command package chooses which of the three each of its
|
||||
commands is added to.
|
||||
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.
|
||||
@@ -13,8 +15,8 @@ commands add new CLI operations, while modules add analysis steps to existing
|
||||
|
||||
## 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 packages can register a Click command or group on one or more of the MVT CLIs.
|
||||
A minimal package can expose this command from `my_mvt_plugin.py`:
|
||||
|
||||
```python
|
||||
import click
|
||||
@@ -38,9 +40,13 @@ summarize = "my_mvt_plugin:summarize"
|
||||
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:
|
||||
Each entry-point group adds the command to one CLI: `mvt.ios.cli_plugins` to
|
||||
`mvt-ios`, `mvt.android.cli_plugins` to `mvt-android` and `mvt.cli_plugins` to
|
||||
`mvt`. Register the command in the group of every CLI which should offer it: a
|
||||
platform-specific command belongs in one platform group, and a command which
|
||||
handles acquisitions of both platforms, as above, in both. After installing the
|
||||
package in the same environment as MVT, the command appears directly in those
|
||||
CLIs:
|
||||
|
||||
```bash
|
||||
mvt-ios summarize ./ios-backup
|
||||
@@ -54,16 +60,39 @@ pipx inject mvt my-mvt-plugin
|
||||
```
|
||||
|
||||
When MVT is installed in an active virtual environment, install the plugin with
|
||||
`pip` in that environment.
|
||||
`pip` in that environment. `mvt plugins list` shows the installed packages and
|
||||
the commands they add, see [Managing Plugins](plugins.md).
|
||||
|
||||
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
|
||||
### Commands on `mvt`
|
||||
|
||||
For local commands that are not packaged, create a Python file that exports one
|
||||
Click command or group named `cli`:
|
||||
The `mvt` command hosts what belongs to neither platform: `version`,
|
||||
`completion`, `plugins` and `download-iocs`. A plugin command which is not
|
||||
about the acquisition of one platform, such as one which configures the plugin
|
||||
or synchronizes the indicators it uses, belongs there too, in the
|
||||
`mvt.cli_plugins` group:
|
||||
|
||||
```toml
|
||||
[project.entry-points."mvt.cli_plugins"]
|
||||
my-plugin = "my_mvt_plugin:my_plugin"
|
||||
```
|
||||
|
||||
Commands in this group are added to `mvt` only, so this one is invoked as
|
||||
`mvt my-plugin`. A command on `mvt` has nothing but its name to say which
|
||||
plugin it belongs to, so name it after the plugin, and make it a Click group
|
||||
when the plugin has several operations to offer, such as
|
||||
`mvt my-plugin configure`.
|
||||
|
||||
## Developing a Command Locally
|
||||
|
||||
A package is how a command is distributed. While a command is being written,
|
||||
MVT can load it straight from its file instead, so the package need not be
|
||||
reinstalled after every change; an editable install of the package does the
|
||||
same through its entry points. Create a Python file that exports one Click
|
||||
command or group named `cli`:
|
||||
|
||||
```python
|
||||
import click
|
||||
@@ -86,10 +115,12 @@ mvt-ios --load-command ./case_summary.py case-summary ./ios-backup
|
||||
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:
|
||||
To load a file or folder on every invocation, set the environment variable of
|
||||
the CLI the commands belong on. Like the entry-point groups, each variable adds
|
||||
its commands to one CLI only:
|
||||
|
||||
```bash
|
||||
export MVT_CUSTOM_COMMANDS=./commands
|
||||
export MVT_IOS_CUSTOM_COMMANDS=./ios_commands
|
||||
export MVT_ANDROID_CUSTOM_COMMANDS=./android_commands
|
||||
```
|
||||
@@ -97,9 +128,11 @@ 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.
|
||||
unique on each CLI; when installed packages or environment paths collide, MVT
|
||||
keeps the first command and logs a warning. The environment path of a CLI is
|
||||
registered before its installed packages, so a command loaded from there wins a
|
||||
collision with a package. 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.
|
||||
|
||||
@@ -214,6 +214,9 @@ module came from:
|
||||
- When a command runs with an `--output` folder, the `command.log` file
|
||||
records one line per module source with the source's version or hash and
|
||||
the list of modules loaded from it.
|
||||
- `mvt plugins list` lists the installed packages, where each of them
|
||||
was installed from and how many modules it contributes, see
|
||||
[Managing Plugins](plugins.md).
|
||||
|
||||
## Profiling
|
||||
|
||||
|
||||
@@ -0,0 +1,116 @@
|
||||
# Managing Plugins
|
||||
|
||||
Plugin packages extend MVT with additional
|
||||
[forensic modules](index.md#custom-modules) and
|
||||
[CLI commands](custom_commands.md). Because installed packages load
|
||||
automatically, `mvt plugins` audits what is installed and checks whether
|
||||
updates are available. The command lives on `mvt` only, although the packages
|
||||
it lists extend `mvt-ios` and `mvt-android` too.
|
||||
|
||||
## List Installed Plugins
|
||||
|
||||
```bash
|
||||
mvt plugins list
|
||||
```
|
||||
|
||||
```
|
||||
Installed MVT plugins
|
||||
┏━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━┳━━━━━━━━━━━━━━┳━━━━━━━━━┳━━━━━━━━━━━┓
|
||||
┃ Name ┃ Version ┃ Origin ┃ Modules ┃ Commands ┃
|
||||
┡━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━╇━━━━━━━━━━━━━━╇━━━━━━━━━╇━━━━━━━━━━━┩
|
||||
│ mvt-plugin-example │ 1.2.0 │ pypi │ 4 │ summarize │
|
||||
│ mvt-plugin-research │ 0.1.0 │ git+3f9a1c7d │ 2 │ - │
|
||||
│ mvt-plugin-local │ 0.0.1 │ local │ 1 │ triage │
|
||||
└─────────────────────┴─────────┴──────────────┴─────────┴───────────┘
|
||||
```
|
||||
|
||||
The origin records where each package was installed from: `pypi` for a package
|
||||
installed from a package index, `git+<commit>` for a package installed directly
|
||||
from a repository, and `local` for a package installed from a local folder or
|
||||
archive rather than from an index, including an editable development install.
|
||||
The last two columns show how many forensic modules the package contributes and
|
||||
which CLI commands it adds.
|
||||
|
||||
A plugin whose modules cannot be imported is listed with `error` in the
|
||||
`Modules` column rather than breaking the listing.
|
||||
|
||||
## Check for Updates
|
||||
|
||||
```bash
|
||||
mvt plugins check-updates
|
||||
```
|
||||
|
||||
```
|
||||
Plugin updates available:
|
||||
mvt-plugin-example 1.2.0 → 1.3.0
|
||||
Upgrade with: pip install -U mvt-plugin-example
|
||||
|
||||
MVT does not install plugin updates. Run the command above when you decide to
|
||||
upgrade.
|
||||
```
|
||||
|
||||
Packages installed from a package index are compared against the latest release
|
||||
published for them. A package which was never published, for example a plugin
|
||||
distributed only within an organization, is skipped silently.
|
||||
|
||||
!!! note
|
||||
|
||||
Packages shown with the `pypi` origin are compared against
|
||||
[PyPI](https://pypi.org), whichever index they were installed from. A
|
||||
plugin installed from a private index under a name which also exists on
|
||||
PyPI is therefore compared against the unrelated public package of that
|
||||
name. Give plugins published to a private index a name which is not taken
|
||||
on PyPI, and treat an unexpected update suggestion as a reason to check
|
||||
where the package would come from.
|
||||
|
||||
!!! warning
|
||||
|
||||
MVT never installs or upgrades a plugin itself, it only prints the command
|
||||
which does. Upgrading a plugin in the middle of an investigation changes
|
||||
the modules producing the results, and a plugin runs as trusted code inside
|
||||
the MVT process, so pulling in a new version is a decision for the analyst
|
||||
to make deliberately and not a side effect of running a check.
|
||||
|
||||
## Automatic Update Checks
|
||||
|
||||
MVT also reports available plugin updates in the banner printed when a command
|
||||
starts:
|
||||
|
||||
```
|
||||
MVT - Mobile Verification Toolkit
|
||||
|
||||
https://mvt.re
|
||||
Version: 2026.7.29
|
||||
|
||||
Plugin updates available:
|
||||
mvt-plugin-example 1.2.0 → 1.3.0 (pip install -U mvt-plugin-example)
|
||||
```
|
||||
|
||||
This check runs at most once every 12 hours. In between checks MVT prints the
|
||||
findings of the latest check without contacting anything, so a plugin update
|
||||
stays visible without a lookup on every command. The
|
||||
`mvt plugins check-updates` command checks immediately, regardless of when the
|
||||
last check happened.
|
||||
|
||||
The automatic check is skipped when the `--disable-update-check` option is
|
||||
used, when `NETWORK_ACCESS_ALLOWED` is disabled in the MVT configuration, and
|
||||
when no plugins are installed.
|
||||
|
||||
## Plugins Installed From a Repository
|
||||
|
||||
A plugin installed with `pip install "mvt-plugin-example @ git+<url>"` is
|
||||
checked by asking the remote repository which commit the installed revision
|
||||
points at now. MVT runs git and ssh in batch mode, so a repository which needs
|
||||
credentials MVT does not already have fails the check instead of prompting for
|
||||
them. The check is skipped silently when git is not available, when the
|
||||
repository cannot be reached, and when access to it is denied.
|
||||
|
||||
How the plugin was installed decides what an update means:
|
||||
|
||||
- A plugin installed from a branch is reported as outdated when the branch has
|
||||
moved past the installed commit.
|
||||
- A plugin installed from a specific commit or a tag is pinned. It is never
|
||||
reported as outdated, however far the branch it came from moves on.
|
||||
|
||||
Pinning a plugin to a commit or a tag is therefore the way to keep the modules
|
||||
used across an investigation stable.
|
||||
+1
-1
@@ -9,7 +9,7 @@ Mobile Verification Toolkit (MVT) is a tool to facilitate the [consensual forens
|
||||
It has been developed and released by the [Amnesty International Security Lab](https://securitylab.amnesty.org) in July 2021 in the context of the [Pegasus Project](https://forbiddenstories.org/about-the-pegasus-project/) along with [a technical forensic methodology](https://www.amnesty.org/en/latest/research/2021/07/forensic-methodology-report-how-to-catch-nso-groups-pegasus/). It continues to be maintained by Amnesty International and other contributors.
|
||||
|
||||
|
||||
In this documentation you will find instructions on how to install and run the `mvt-ios` and `mvt-android` commands, and guidance on how to interpret the extracted results.
|
||||
In this documentation you will find instructions on how to install and run the `mvt-ios`, `mvt-android` and `mvt` commands, and guidance on how to interpret the extracted results.
|
||||
|
||||
## Resources
|
||||
|
||||
|
||||
+3
-3
@@ -64,7 +64,7 @@ It is recommended to try installing and running MVT from [Windows Subsystem Linu
|
||||
pipx install mvt
|
||||
```
|
||||
|
||||
You now should have the `mvt-ios` and `mvt-android` utilities installed. If you run into problems with these commands not being found, ensure you have run `pipx ensurepath` and opened a new terminal window.
|
||||
You now should have the `mvt`, `mvt-ios` and `mvt-android` utilities installed. If you run into problems with these commands not being found, ensure you have run `pipx ensurepath` and opened a new terminal window.
|
||||
|
||||
### Installing from PyPI directly into a virtual environment
|
||||
You can use `pipenv`, `poetry` etc. for your virtual environment, but the provided example is with the built-in `venv` tool:
|
||||
@@ -84,7 +84,7 @@ source env/bin/activate
|
||||
pip install mvt
|
||||
```
|
||||
|
||||
The `mvt-ios` and `mvt-android` utilities should now be available as commands whenever the virtual environment is active.
|
||||
The `mvt`, `mvt-ios` and `mvt-android` utilities should now be available as commands whenever the virtual environment is active.
|
||||
|
||||
### Installing from git source with pipx
|
||||
If you want to have the latest features in development, you can install MVT directly from the source code in git.
|
||||
@@ -93,7 +93,7 @@ If you want to have the latest features in development, you can install MVT dire
|
||||
pipx install --force git+https://github.com/mvt-project/mvt.git
|
||||
```
|
||||
|
||||
You now should have the `mvt-ios` and `mvt-android` utilities installed.
|
||||
You now should have the `mvt`, `mvt-ios` and `mvt-android` utilities installed.
|
||||
|
||||
**Notes:**
|
||||
1. The `--force` flag is necessary to force the reinstallation of the package.
|
||||
|
||||
+1
-1
@@ -71,7 +71,7 @@ So far MVT implements only a subset of [STIX2 specifications](https://docs.oasis
|
||||
- [This repository](https://github.com/Te-k/stalkerware-indicators) contains IOCs for Android stalkerware including [a STIX MVT-compatible file](https://raw.githubusercontent.com/Te-k/stalkerware-indicators/master/generated/stalkerware.stix2).
|
||||
- We are also maintaining [a list of IOCs](https://github.com/mvt-project/mvt-indicators) in STIX format from public spyware campaigns.
|
||||
|
||||
You can automaticallly download the latest public indicator files with the command `mvt-ios download-iocs` or `mvt-android download-iocs`. These commands download the list of indicators from the [mvt-indicators](https://github.com/mvt-project/mvt-indicators/blob/main/indicators.yaml) repository and store them in the [appdir](https://pypi.org/project/appdirs/) folder. They are then loaded automatically by MVT.
|
||||
You can automatically download the latest public indicator files with the command `mvt download-iocs`. The per-platform forms `mvt-ios download-iocs` and `mvt-android download-iocs` do the same thing. These commands download the list of indicators from the [mvt-indicators](https://github.com/mvt-project/mvt-indicators/blob/main/indicators.yaml) repository and store them in the [appdir](https://pypi.org/project/appdirs/) folder. They are then loaded automatically by MVT.
|
||||
|
||||
Please [open an issue](https://github.com/mvt-project/mvt/issues/) to suggest new sources of STIX-formatted IOCs.
|
||||
|
||||
@@ -53,4 +53,5 @@ nav:
|
||||
- Development Instructions: "development/index.md"
|
||||
- Custom CLI Commands: "development/custom_commands.md"
|
||||
- Plugin Configuration: "development/plugin_configuration.md"
|
||||
- Managing Plugins: "development/plugins.md"
|
||||
- License: "license.md"
|
||||
@@ -44,6 +44,7 @@ homepage = "https://docs.mvt.re/en/latest/"
|
||||
repository = "https://github.com/mvt-project/mvt"
|
||||
|
||||
[project.scripts]
|
||||
mvt = "mvt.cli:main"
|
||||
mvt-ios = "mvt.ios:main"
|
||||
mvt-android = "mvt.android:main"
|
||||
|
||||
|
||||
+5
-47
@@ -15,12 +15,6 @@ from mvt.common.cli_plugins import (
|
||||
register_cli_plugins,
|
||||
)
|
||||
from mvt.common.cmd_check_iocs import CmdCheckIOCS
|
||||
from mvt.common.completion import (
|
||||
SUPPORTED_SHELLS,
|
||||
completion_instructions,
|
||||
generate_completion_script,
|
||||
install_completion_script,
|
||||
)
|
||||
from mvt.common.help import (
|
||||
HELP_MSG_ANDROID_BACKUP_PASSWORD,
|
||||
HELP_MSG_CHECK_ADB_REMOVED,
|
||||
@@ -31,7 +25,6 @@ from mvt.common.help import (
|
||||
HELP_MSG_CHECK_IOCS,
|
||||
HELP_MSG_CHECK_INTRUSION_LOGS,
|
||||
HELP_MSG_DELAY_CHECKS,
|
||||
HELP_MSG_COMPLETION,
|
||||
HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK,
|
||||
HELP_MSG_DISABLE_UPDATE_CHECK,
|
||||
HELP_MSG_HASHES,
|
||||
@@ -102,55 +95,20 @@ def cli(ctx, disable_update_check, disable_indicator_update_check):
|
||||
ctx.ensure_object(dict)
|
||||
ctx.obj["disable_version_check"] = disable_update_check
|
||||
ctx.obj["disable_indicator_check"] = disable_indicator_update_check
|
||||
if ctx.invoked_subcommand != "completion":
|
||||
logo(
|
||||
disable_version_check=disable_update_check,
|
||||
disable_indicator_check=disable_indicator_update_check,
|
||||
)
|
||||
logo(
|
||||
disable_version_check=disable_update_check,
|
||||
disable_indicator_check=disable_indicator_update_check,
|
||||
)
|
||||
|
||||
|
||||
# ==============================================================================
|
||||
# Command: version
|
||||
# ==============================================================================
|
||||
@cli.command("version", help=HELP_MSG_VERSION)
|
||||
@cli.command("version", context_settings=CONTEXT_SETTINGS, help=HELP_MSG_VERSION)
|
||||
def version():
|
||||
return
|
||||
|
||||
|
||||
# ==============================================================================
|
||||
# Command: completion
|
||||
# ==============================================================================
|
||||
@cli.command("completion", context_settings=CONTEXT_SETTINGS, help=HELP_MSG_COMPLETION)
|
||||
@click.argument("shell", required=False, type=click.Choice(SUPPORTED_SHELLS))
|
||||
@click.option(
|
||||
"--install",
|
||||
is_flag=True,
|
||||
help="Write completion files and update shell configuration.",
|
||||
)
|
||||
@click.pass_context
|
||||
def completion(ctx, shell, install):
|
||||
program_name = "mvt-android"
|
||||
|
||||
if shell is None:
|
||||
if install:
|
||||
raise click.UsageError("A shell is required when using --install.")
|
||||
click.echo(completion_instructions(program_name))
|
||||
return
|
||||
|
||||
root_cli = ctx.find_root().command
|
||||
|
||||
if install:
|
||||
script_path = install_completion_script(root_cli, program_name, shell)
|
||||
click.echo(f"Installed {shell} completion to {script_path}")
|
||||
if shell in ("bash", "zsh"):
|
||||
click.echo(f"Updated ~/.{shell}rc")
|
||||
else:
|
||||
click.echo("Fish loads completion files automatically.")
|
||||
return
|
||||
|
||||
click.echo(generate_completion_script(root_cli, program_name, shell))
|
||||
|
||||
|
||||
# ==============================================================================
|
||||
# Command: check-adb (removed)
|
||||
# ==============================================================================
|
||||
|
||||
+108
@@ -0,0 +1,108 @@
|
||||
# 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 click
|
||||
|
||||
from mvt.common.cli_plugins import (
|
||||
MVT_CUSTOM_COMMANDS_ENV,
|
||||
NEUTRAL_CLI_PLUGIN_GROUP,
|
||||
load_cli_commands_option,
|
||||
register_cli_plugins,
|
||||
)
|
||||
from mvt.common.cmd_plugins import plugins
|
||||
from mvt.common.completion import completion
|
||||
from mvt.common.help import (
|
||||
HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK,
|
||||
HELP_MSG_DISABLE_UPDATE_CHECK,
|
||||
HELP_MSG_STIX2,
|
||||
HELP_MSG_VERSION,
|
||||
)
|
||||
from mvt.common.logo import logo
|
||||
from mvt.common.updates import IndicatorsUpdates
|
||||
from mvt.common.utils import init_logging
|
||||
|
||||
init_logging()
|
||||
|
||||
CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])
|
||||
|
||||
|
||||
# ==============================================================================
|
||||
# Main
|
||||
# ==============================================================================
|
||||
@click.group(invoke_without_command=True)
|
||||
@load_cli_commands_option
|
||||
@click.option(
|
||||
"--disable-update-check", is_flag=True, help=HELP_MSG_DISABLE_UPDATE_CHECK
|
||||
)
|
||||
@click.option(
|
||||
"--disable-indicator-update-check",
|
||||
is_flag=True,
|
||||
help=HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK,
|
||||
)
|
||||
@click.pass_context
|
||||
def cli(ctx, disable_update_check, disable_indicator_update_check):
|
||||
"""Mobile Verification Toolkit.
|
||||
|
||||
mvt-ios and mvt-android run the forensic analysis of an acquisition: each
|
||||
provides the check-* commands of its platform. This command hosts what
|
||||
belongs to neither platform; run it without a command to see the installed
|
||||
version and the list of what it offers.
|
||||
"""
|
||||
ctx.ensure_object(dict)
|
||||
ctx.obj["disable_version_check"] = disable_update_check
|
||||
ctx.obj["disable_indicator_check"] = disable_indicator_update_check
|
||||
if ctx.invoked_subcommand != "completion":
|
||||
logo(
|
||||
disable_version_check=disable_update_check,
|
||||
disable_indicator_check=disable_indicator_update_check,
|
||||
)
|
||||
if ctx.invoked_subcommand is None:
|
||||
click.echo(ctx.get_help())
|
||||
|
||||
|
||||
# ==============================================================================
|
||||
# Command: download-iocs
|
||||
# ==============================================================================
|
||||
@cli.command("download-iocs", context_settings=CONTEXT_SETTINGS, help=HELP_MSG_STIX2)
|
||||
def download_iocs():
|
||||
ioc_updates = IndicatorsUpdates()
|
||||
ioc_updates.update()
|
||||
|
||||
|
||||
# ==============================================================================
|
||||
# Command: completion
|
||||
# ==============================================================================
|
||||
cli.add_command(completion)
|
||||
|
||||
|
||||
# ==============================================================================
|
||||
# Command: version
|
||||
# ==============================================================================
|
||||
@cli.command("version", context_settings=CONTEXT_SETTINGS, help=HELP_MSG_VERSION)
|
||||
def version():
|
||||
return
|
||||
|
||||
|
||||
# The plugins command is registered as a built-in command, before any external
|
||||
# command, so that an installed package can never replace it.
|
||||
cli.add_command(plugins)
|
||||
|
||||
|
||||
# ==============================================================================
|
||||
# Entry point of the mvt console script
|
||||
# ==============================================================================
|
||||
def main() -> None:
|
||||
"""Register the external commands and run the mvt 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=NEUTRAL_CLI_PLUGIN_GROUP,
|
||||
environment_variable=MVT_CUSTOM_COMMANDS_ENV,
|
||||
)
|
||||
cli()
|
||||
@@ -17,6 +17,9 @@ import click
|
||||
|
||||
IOS_CLI_PLUGIN_GROUP = "mvt.ios.cli_plugins"
|
||||
ANDROID_CLI_PLUGIN_GROUP = "mvt.android.cli_plugins"
|
||||
# Commands in this group are registered on the platform-neutral mvt command only.
|
||||
NEUTRAL_CLI_PLUGIN_GROUP = "mvt.cli_plugins"
|
||||
MVT_CUSTOM_COMMANDS_ENV = "MVT_CUSTOM_COMMANDS"
|
||||
MVT_IOS_CUSTOM_COMMANDS_ENV = "MVT_IOS_CUSTOM_COMMANDS"
|
||||
MVT_ANDROID_CUSTOM_COMMANDS_ENV = "MVT_ANDROID_CUSTOM_COMMANDS"
|
||||
|
||||
@@ -255,6 +258,16 @@ def register_cli_plugins(
|
||||
entry_point_group: str,
|
||||
environment_variable: str,
|
||||
) -> None:
|
||||
"""Register the external commands of one CLI on its group.
|
||||
|
||||
Each CLI has one entry-point group and one environment variable of its
|
||||
own, so a command package chooses the CLIs its commands are added to.
|
||||
|
||||
:param group: CLI group to register the external commands on.
|
||||
:param entry_point_group: Entry-point group of the CLI.
|
||||
:param environment_variable: Name of the environment variable holding a
|
||||
path to load commands from.
|
||||
"""
|
||||
environment_path = os.environ.get(environment_variable)
|
||||
if environment_path:
|
||||
register_cli_commands_from_path(group, environment_path)
|
||||
|
||||
@@ -0,0 +1,205 @@
|
||||
# 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 importlib.metadata
|
||||
import logging
|
||||
from typing import Optional
|
||||
|
||||
import click
|
||||
from rich.console import Console
|
||||
from rich.table import Table
|
||||
|
||||
from .cli_plugins import (
|
||||
ANDROID_CLI_PLUGIN_GROUP,
|
||||
IOS_CLI_PLUGIN_GROUP,
|
||||
NEUTRAL_CLI_PLUGIN_GROUP,
|
||||
)
|
||||
from .config import settings
|
||||
from .help import (
|
||||
HELP_MSG_PLUGINS,
|
||||
HELP_MSG_PLUGINS_CHECK_UPDATES,
|
||||
HELP_MSG_PLUGINS_LIST,
|
||||
)
|
||||
from .module import MVTModule
|
||||
from .module_loader import MODULES_ENTRY_POINT_GROUP, distribution_direct_url
|
||||
from .updates import (
|
||||
SHORT_COMMIT_LENGTH,
|
||||
PluginUpdates,
|
||||
installed_plugin_distributions,
|
||||
)
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
CLI_PLUGIN_GROUPS = (
|
||||
IOS_CLI_PLUGIN_GROUP,
|
||||
ANDROID_CLI_PLUGIN_GROUP,
|
||||
NEUTRAL_CLI_PLUGIN_GROUP,
|
||||
)
|
||||
CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])
|
||||
|
||||
|
||||
def _entry_points(group: str) -> list[importlib.metadata.EntryPoint]:
|
||||
try:
|
||||
return list(importlib.metadata.entry_points(group=group))
|
||||
except Exception as exc:
|
||||
log.warning("Unable to discover the entry points in group %s: %s", group, exc)
|
||||
return []
|
||||
|
||||
|
||||
def _entry_point_distribution(
|
||||
entry_point: importlib.metadata.EntryPoint,
|
||||
) -> Optional[str]:
|
||||
dist = getattr(entry_point, "dist", None)
|
||||
if dist is None:
|
||||
return None
|
||||
try:
|
||||
return dist.name
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def _distribution_version(dist: importlib.metadata.Distribution) -> str:
|
||||
try:
|
||||
return dist.version or "unknown"
|
||||
except Exception:
|
||||
return "unknown"
|
||||
|
||||
|
||||
def _distribution_origin(dist: importlib.metadata.Distribution) -> str:
|
||||
"""Describe where a plugin package was installed from."""
|
||||
direct_url = distribution_direct_url(dist)
|
||||
if direct_url is None:
|
||||
return "pypi"
|
||||
|
||||
vcs_info = direct_url.get("vcs_info")
|
||||
if isinstance(vcs_info, dict):
|
||||
commit = vcs_info.get("commit_id") or ""
|
||||
if commit:
|
||||
return f"git+{commit[:SHORT_COMMIT_LENGTH]}"
|
||||
return "git"
|
||||
|
||||
return "local"
|
||||
|
||||
|
||||
def _contributed_modules(
|
||||
entry_points: list[importlib.metadata.EntryPoint], distribution: str
|
||||
) -> str:
|
||||
"""Count the forensic modules a plugin package contributes.
|
||||
|
||||
Entry points are resolved the way MVT resolves them when it loads
|
||||
modules, but a broken entry point is reported instead of raising: listing
|
||||
the installed plugins must work even when one of them is faulty.
|
||||
"""
|
||||
count = 0
|
||||
broken = False
|
||||
|
||||
for entry_point in entry_points:
|
||||
if _entry_point_distribution(entry_point) != distribution:
|
||||
continue
|
||||
try:
|
||||
loaded = entry_point.load()
|
||||
if callable(loaded) and not isinstance(loaded, type):
|
||||
loaded = loaded()
|
||||
count += sum(
|
||||
1
|
||||
for module in loaded
|
||||
if isinstance(module, type) and issubclass(module, MVTModule)
|
||||
)
|
||||
except (Exception, SystemExit) as exc:
|
||||
log.debug(
|
||||
"Unable to load the modules of entry point %s (%s): %s",
|
||||
entry_point.name,
|
||||
entry_point.value,
|
||||
exc,
|
||||
)
|
||||
broken = True
|
||||
|
||||
if broken:
|
||||
return f"{count} (error)" if count else "error"
|
||||
|
||||
return str(count)
|
||||
|
||||
|
||||
def _contributed_commands(
|
||||
entry_points: list[importlib.metadata.EntryPoint], distribution: str
|
||||
) -> str:
|
||||
names = {
|
||||
entry_point.name
|
||||
for entry_point in entry_points
|
||||
if _entry_point_distribution(entry_point) == distribution
|
||||
}
|
||||
|
||||
return ", ".join(sorted(names)) if names else "-"
|
||||
|
||||
|
||||
@click.group("plugins", context_settings=CONTEXT_SETTINGS, help=HELP_MSG_PLUGINS)
|
||||
def plugins() -> None:
|
||||
pass
|
||||
|
||||
|
||||
@plugins.command("list", context_settings=CONTEXT_SETTINGS, help=HELP_MSG_PLUGINS_LIST)
|
||||
def list_plugins() -> None:
|
||||
distributions = installed_plugin_distributions()
|
||||
if not distributions:
|
||||
click.echo("No MVT plugins are installed.")
|
||||
return
|
||||
|
||||
module_entry_points = _entry_points(MODULES_ENTRY_POINT_GROUP)
|
||||
command_entry_points = []
|
||||
for group in CLI_PLUGIN_GROUPS:
|
||||
command_entry_points.extend(_entry_points(group))
|
||||
|
||||
table = Table(title="Installed MVT plugins")
|
||||
table.add_column("Name", style="bold")
|
||||
table.add_column("Version")
|
||||
table.add_column("Origin")
|
||||
table.add_column("Modules", justify="right")
|
||||
table.add_column("Commands")
|
||||
|
||||
for dist in distributions:
|
||||
name = dist.name
|
||||
table.add_row(
|
||||
name,
|
||||
_distribution_version(dist),
|
||||
_distribution_origin(dist),
|
||||
_contributed_modules(module_entry_points, name),
|
||||
_contributed_commands(command_entry_points, name),
|
||||
)
|
||||
|
||||
Console().print(table)
|
||||
|
||||
|
||||
@plugins.command(
|
||||
"check-updates",
|
||||
context_settings=CONTEXT_SETTINGS,
|
||||
help=HELP_MSG_PLUGINS_CHECK_UPDATES,
|
||||
short_help="Check the installed plugins for updates",
|
||||
)
|
||||
def check_plugin_updates() -> None:
|
||||
if not settings.NETWORK_ACCESS_ALLOWED:
|
||||
click.echo(
|
||||
"Network access is disabled, cannot check for plugin updates. "
|
||||
"Enable NETWORK_ACCESS_ALLOWED in the MVT configuration to check."
|
||||
)
|
||||
return
|
||||
|
||||
if not installed_plugin_distributions():
|
||||
click.echo("No MVT plugins are installed.")
|
||||
return
|
||||
|
||||
findings = PluginUpdates().check()
|
||||
if not findings:
|
||||
click.echo("All plugins are up to date.")
|
||||
return
|
||||
|
||||
click.echo("Plugin updates available:")
|
||||
for finding in findings:
|
||||
click.echo(f" {finding['name']} {finding['installed']} → {finding['latest']}")
|
||||
click.echo(f" Upgrade with: {finding['upgrade_command']}")
|
||||
|
||||
click.echo(
|
||||
"\nMVT does not install plugin updates. Run the command above when you "
|
||||
"decide to upgrade."
|
||||
)
|
||||
@@ -9,34 +9,83 @@ import shlex
|
||||
import click
|
||||
from click.shell_completion import get_completion_class
|
||||
|
||||
from .help import HELP_MSG_COMPLETION
|
||||
|
||||
|
||||
SUPPORTED_SHELLS = ("bash", "zsh", "fish")
|
||||
CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])
|
||||
|
||||
COMPLETION_INSTRUCTIONS = """Shell completion for mvt, mvt-ios and mvt-android
|
||||
|
||||
def completion_instructions(program_name: str) -> str:
|
||||
return f"""Shell completion for {program_name}
|
||||
|
||||
Print a completion script:
|
||||
{program_name} completion bash > ~/.{program_name}-complete.bash
|
||||
{program_name} completion zsh > ~/.{program_name}-complete.zsh
|
||||
mkdir -p ~/.config/fish/completions
|
||||
{program_name} completion fish > ~/.config/fish/completions/{program_name}.fish
|
||||
Print one completion script covering the three commands:
|
||||
mvt completion bash > ~/.mvt-complete.bash
|
||||
mvt completion zsh > ~/.mvt-complete.zsh
|
||||
mkdir -p ~/.config/fish/conf.d
|
||||
mvt completion fish > ~/.config/fish/conf.d/mvt-completion.fish
|
||||
|
||||
Load the generated Bash script from ~/.bashrc:
|
||||
[ -f ~/.{program_name}-complete.bash ] && . ~/.{program_name}-complete.bash
|
||||
[ -f ~/.mvt-complete.bash ] && . ~/.mvt-complete.bash
|
||||
|
||||
Load the generated Zsh script from ~/.zshrc:
|
||||
[ -f ~/.{program_name}-complete.zsh ] && . ~/.{program_name}-complete.zsh
|
||||
[ -f ~/.mvt-complete.zsh ] && . ~/.mvt-complete.zsh
|
||||
|
||||
Fish loads completion files from ~/.config/fish/completions automatically.
|
||||
Fish loads the files in ~/.config/fish/conf.d automatically.
|
||||
|
||||
To write these files and update Bash/Zsh shell configuration automatically:
|
||||
{program_name} completion bash --install
|
||||
{program_name} completion zsh --install
|
||||
{program_name} completion fish --install
|
||||
To write these files and update the Bash/Zsh shell configuration automatically:
|
||||
mvt completion bash --install
|
||||
mvt completion zsh --install
|
||||
mvt completion fish --install
|
||||
"""
|
||||
|
||||
|
||||
def _mvt_programs() -> list[tuple[str, click.Command]]:
|
||||
"""Return the console script name and CLI group of every MVT program.
|
||||
|
||||
The three CLIs are imported here rather than at module level: mvt.cli
|
||||
imports this module while it is being defined, and generating a completion
|
||||
script should not make the start-up of `mvt` import the platform CLIs.
|
||||
"""
|
||||
from mvt.android.cli import cli as android_cli
|
||||
from mvt.cli import cli as mvt_cli
|
||||
from mvt.ios.cli import cli as ios_cli
|
||||
|
||||
return [("mvt", mvt_cli), ("mvt-ios", ios_cli), ("mvt-android", android_cli)]
|
||||
|
||||
|
||||
@click.command(
|
||||
"completion",
|
||||
context_settings=CONTEXT_SETTINGS,
|
||||
help=HELP_MSG_COMPLETION,
|
||||
short_help="Generate or install shell completion",
|
||||
)
|
||||
@click.argument("shell", required=False, type=click.Choice(SUPPORTED_SHELLS))
|
||||
@click.option(
|
||||
"--install",
|
||||
is_flag=True,
|
||||
help="Write completion files and update shell configuration.",
|
||||
)
|
||||
def completion(shell, install):
|
||||
if shell is None:
|
||||
if install:
|
||||
raise click.UsageError("A shell is required when using --install.")
|
||||
click.echo(COMPLETION_INSTRUCTIONS)
|
||||
return
|
||||
|
||||
if install:
|
||||
script_path = install_completion_script(shell)
|
||||
click.echo(
|
||||
f"Installed {shell} completion for mvt, mvt-ios and mvt-android "
|
||||
f"to {script_path}"
|
||||
)
|
||||
if shell in ("bash", "zsh"):
|
||||
click.echo(f"Updated ~/.{shell}rc")
|
||||
else:
|
||||
click.echo("Fish loads the files in ~/.config/fish/conf.d automatically.")
|
||||
return
|
||||
|
||||
click.echo(generate_mvt_completion_script(shell))
|
||||
|
||||
|
||||
def generate_completion_script(cli: click.Command, program_name: str, shell: str) -> str:
|
||||
completion_class = get_completion_class(shell)
|
||||
if completion_class is None:
|
||||
@@ -46,41 +95,49 @@ def generate_completion_script(cli: click.Command, program_name: str, shell: str
|
||||
return completion_class(cli, {}, program_name, complete_var).source()
|
||||
|
||||
|
||||
def install_completion_script(
|
||||
cli: click.Command,
|
||||
program_name: str,
|
||||
shell: str,
|
||||
) -> Path:
|
||||
script = generate_completion_script(cli, program_name, shell)
|
||||
script_path = _completion_script_path(program_name, shell)
|
||||
def generate_mvt_completion_script(shell: str) -> str:
|
||||
"""Return one script completing every MVT command.
|
||||
|
||||
Click names the completion function of each program after the program, so
|
||||
the scripts of the three commands can simply be concatenated.
|
||||
"""
|
||||
scripts = [
|
||||
generate_completion_script(cli, program_name, shell).strip("\n")
|
||||
for program_name, cli in _mvt_programs()
|
||||
]
|
||||
return "\n\n".join(scripts)
|
||||
|
||||
|
||||
def install_completion_script(shell: str) -> Path:
|
||||
script = generate_mvt_completion_script(shell)
|
||||
script_path = _completion_script_path(shell)
|
||||
script_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
script_path.write_text(script, encoding="utf-8")
|
||||
script_path.write_text(f"{script}\n", encoding="utf-8")
|
||||
|
||||
if shell in ("bash", "zsh"):
|
||||
_install_shell_source_line(program_name, shell, script_path)
|
||||
_install_shell_source_line(shell, script_path)
|
||||
|
||||
return script_path
|
||||
|
||||
|
||||
def _completion_script_path(program_name: str, shell: str) -> Path:
|
||||
def _completion_script_path(shell: str) -> Path:
|
||||
home = Path.home()
|
||||
|
||||
if shell == "fish":
|
||||
return home / ".config" / "fish" / "completions" / f"{program_name}.fish"
|
||||
# conf.d is sourced when the shell starts, unlike the completions
|
||||
# folder, whose files fish loads on demand by command name.
|
||||
return home / ".config" / "fish" / "conf.d" / "mvt-completion.fish"
|
||||
|
||||
return home / f".{program_name}-complete.{shell}"
|
||||
return home / f".mvt-complete.{shell}"
|
||||
|
||||
|
||||
def _install_shell_source_line(program_name: str, shell: str, script_path: Path) -> None:
|
||||
def _install_shell_source_line(shell: str, script_path: Path) -> None:
|
||||
shell_config_path = Path.home() / f".{shell}rc"
|
||||
source_line = (
|
||||
f"[ -f {shlex.quote(str(script_path))} ] && "
|
||||
f". {shlex.quote(str(script_path))}"
|
||||
)
|
||||
block = (
|
||||
f"# MVT shell completion for {program_name}\n"
|
||||
f"{source_line}\n"
|
||||
)
|
||||
block = f"# MVT shell completion\n{source_line}\n"
|
||||
|
||||
if shell_config_path.exists():
|
||||
shell_config = shell_config_path.read_text(encoding="utf-8")
|
||||
|
||||
@@ -21,7 +21,14 @@ HELP_MSG_CHECK_IOCS = "Compare stored JSON results to provided indicators"
|
||||
HELP_MSG_STIX2 = "Download public STIX2 indicators"
|
||||
HELP_MSG_DISABLE_UPDATE_CHECK = "Disable MVT version update check"
|
||||
HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK = "Disable indicators update check"
|
||||
HELP_MSG_COMPLETION = "Generate or install shell completion"
|
||||
HELP_MSG_COMPLETION = (
|
||||
"Generate or install shell completion for mvt, mvt-ios and mvt-android"
|
||||
)
|
||||
HELP_MSG_PLUGINS = "Inspect the installed MVT plugin packages"
|
||||
HELP_MSG_PLUGINS_LIST = "List the installed plugins and what they contribute to MVT"
|
||||
HELP_MSG_PLUGINS_CHECK_UPDATES = (
|
||||
"Check the installed plugins for updates without installing them"
|
||||
)
|
||||
|
||||
# IOS Specific
|
||||
HELP_MSG_DECRYPT_BACKUP = "Decrypt an encrypted iTunes backup"
|
||||
|
||||
+125
-62
@@ -8,10 +8,125 @@ import logging
|
||||
import requests
|
||||
from rich import print as rich_print
|
||||
|
||||
from .updates import IndicatorsUpdates, MVTUpdates
|
||||
from .config import settings
|
||||
from .updates import (
|
||||
IndicatorsUpdates,
|
||||
MVTUpdates,
|
||||
PluginUpdates,
|
||||
installed_plugin_distributions,
|
||||
)
|
||||
from .version import MVT_VERSION
|
||||
|
||||
|
||||
def _check_version_updates(log: logging.Logger) -> None:
|
||||
try:
|
||||
mvt_updates = MVTUpdates()
|
||||
latest_version = mvt_updates.check()
|
||||
except (requests.exceptions.ConnectionError, requests.exceptions.Timeout):
|
||||
rich_print(
|
||||
"\t[bold]Note: Could not check for MVT updates.[/bold] "
|
||||
"You may be working offline. Please update MVT regularly."
|
||||
)
|
||||
except Exception as e:
|
||||
log.error("Error encountered when trying to check latest MVT version: %s", e)
|
||||
else:
|
||||
if latest_version:
|
||||
rich_print(
|
||||
f"\t[bold]Version {latest_version} is available! "
|
||||
"Upgrade mvt with `pip3 install -U mvt` or with `pipx upgrade mvt`[/bold]"
|
||||
)
|
||||
|
||||
|
||||
def _check_indicator_updates(log: logging.Logger) -> None:
|
||||
ioc_updates = IndicatorsUpdates()
|
||||
|
||||
# Before proceeding, we check if we have downloaded an indicators index.
|
||||
# If not, there's no point in proceeding with the updates check.
|
||||
if ioc_updates.get_latest_update() == 0:
|
||||
rich_print(
|
||||
"\t[bold]You have not yet downloaded any indicators, check "
|
||||
"the `download-iocs` command![/bold]"
|
||||
)
|
||||
return
|
||||
|
||||
# We only perform this check at a fixed frequency, in order to not
|
||||
# overburden the user with too many lookups if the command is being run
|
||||
# multiple times.
|
||||
should_check, hours = ioc_updates.should_check()
|
||||
if not should_check:
|
||||
rich_print(
|
||||
f"\tIndicators updates checked recently, next automatic check "
|
||||
f"in {int(hours)} hours"
|
||||
)
|
||||
return
|
||||
|
||||
try:
|
||||
ioc_to_update = ioc_updates.check()
|
||||
except (requests.exceptions.ConnectionError, requests.exceptions.Timeout):
|
||||
rich_print(
|
||||
"\t[bold]Note: Could not check for indicator updates.[/bold] "
|
||||
"You may be working offline. Please update MVT indicators regularly."
|
||||
)
|
||||
except Exception as e:
|
||||
log.error("Error encountered when trying to check latest MVT indicators: %s", e)
|
||||
else:
|
||||
if ioc_to_update:
|
||||
rich_print(
|
||||
"\t[bold]There are updates to your indicators files! "
|
||||
"Run the `download-iocs` command to update![/bold]"
|
||||
)
|
||||
else:
|
||||
rich_print("\tYour indicators files seem to be up to date.")
|
||||
|
||||
|
||||
def _print_plugin_updates(findings: list) -> None:
|
||||
if not findings:
|
||||
return
|
||||
|
||||
rich_print("\t[bold]Plugin updates available:[/bold]")
|
||||
for finding in findings:
|
||||
rich_print(
|
||||
f"\t {finding['name']} {finding['installed']} → "
|
||||
f"{finding['latest']} ({finding['upgrade_command']})"
|
||||
)
|
||||
|
||||
|
||||
def _check_plugin_updates(log: logging.Logger) -> None:
|
||||
if not settings.NETWORK_ACCESS_ALLOWED:
|
||||
return
|
||||
|
||||
# This runs on every command, so nothing here, including reading back what
|
||||
# the latest check stored, may ever interrupt MVT.
|
||||
try:
|
||||
distributions = installed_plugin_distributions()
|
||||
|
||||
# There is nothing to check when MVT was not extended with any plugin.
|
||||
if not distributions:
|
||||
return
|
||||
|
||||
plugin_updates = PluginUpdates()
|
||||
|
||||
# We only perform this check at a fixed frequency, in order to not
|
||||
# overburden the user (and the plugin repositories) with too many
|
||||
# lookups. In between checks we print the findings of the latest one,
|
||||
# leaving out those which no longer apply to what is installed.
|
||||
should_check, _ = plugin_updates.should_check()
|
||||
if not should_check:
|
||||
_print_plugin_updates(plugin_updates.current_findings(distributions))
|
||||
return
|
||||
|
||||
findings = plugin_updates.check()
|
||||
except (requests.exceptions.ConnectionError, requests.exceptions.Timeout):
|
||||
rich_print(
|
||||
"\t[bold]Note: Could not check for plugin updates.[/bold] "
|
||||
"You may be working offline. Please update your MVT plugins regularly."
|
||||
)
|
||||
except Exception as e:
|
||||
log.error("Error encountered when trying to check MVT plugin updates: %s", e)
|
||||
else:
|
||||
_print_plugin_updates(findings)
|
||||
|
||||
|
||||
def check_updates(
|
||||
disable_version_check: bool = False, disable_indicator_check: bool = False
|
||||
) -> None:
|
||||
@@ -19,77 +134,25 @@ def check_updates(
|
||||
|
||||
# First we check for MVT version updates.
|
||||
if not disable_version_check:
|
||||
try:
|
||||
mvt_updates = MVTUpdates()
|
||||
latest_version = mvt_updates.check()
|
||||
except (requests.exceptions.ConnectionError, requests.exceptions.Timeout):
|
||||
rich_print(
|
||||
"\t\t[bold]Note: Could not check for MVT updates.[/bold] "
|
||||
"You may be working offline. Please update MVT regularly."
|
||||
)
|
||||
except Exception as e:
|
||||
log.error(
|
||||
"Error encountered when trying to check latest MVT version: %s", e
|
||||
)
|
||||
else:
|
||||
if latest_version:
|
||||
rich_print(
|
||||
f"\t\t[bold]Version {latest_version} is available! "
|
||||
"Upgrade mvt with `pip3 install -U mvt` or with `pipx upgrade mvt`[/bold]"
|
||||
)
|
||||
_check_version_updates(log)
|
||||
|
||||
# Then we check for indicators files updates.
|
||||
if not disable_indicator_check:
|
||||
ioc_updates = IndicatorsUpdates()
|
||||
_check_indicator_updates(log)
|
||||
|
||||
# Before proceeding, we check if we have downloaded an indicators index.
|
||||
# If not, there's no point in proceeding with the updates check.
|
||||
if ioc_updates.get_latest_update() == 0:
|
||||
rich_print(
|
||||
"\t\t[bold]You have not yet downloaded any indicators, check "
|
||||
"the `download-iocs` command![/bold]"
|
||||
)
|
||||
return
|
||||
|
||||
# We only perform this check at a fixed frequency, in order to not
|
||||
# overburden the user with too many lookups if the command is being run
|
||||
# multiple times.
|
||||
should_check, hours = ioc_updates.should_check()
|
||||
if not should_check:
|
||||
rich_print(
|
||||
f"\t\tIndicators updates checked recently, next automatic check "
|
||||
f"in {int(hours)} hours"
|
||||
)
|
||||
return
|
||||
|
||||
try:
|
||||
ioc_to_update = ioc_updates.check()
|
||||
except (requests.exceptions.ConnectionError, requests.exceptions.Timeout):
|
||||
rich_print(
|
||||
"\t\t[bold]Note: Could not check for indicator updates.[/bold] "
|
||||
"You may be working offline. Please update MVT indicators regularly."
|
||||
)
|
||||
except Exception as e:
|
||||
log.error(
|
||||
"Error encountered when trying to check latest MVT indicators: %s", e
|
||||
)
|
||||
else:
|
||||
if ioc_to_update:
|
||||
rich_print(
|
||||
"\t\t[bold]There are updates to your indicators files! "
|
||||
"Run the `download-iocs` command to update![/bold]"
|
||||
)
|
||||
else:
|
||||
rich_print("\t\tYour indicators files seem to be up to date.")
|
||||
# Finally we check for updates to the installed plugin packages. MVT never
|
||||
# installs an update itself, it only reports the command which does.
|
||||
if not disable_version_check:
|
||||
_check_plugin_updates(log)
|
||||
|
||||
|
||||
def logo(
|
||||
disable_version_check: bool = False, disable_indicator_check: bool = False
|
||||
) -> None:
|
||||
rich_print("\n")
|
||||
rich_print("\t[bold]MVT[/bold] - Mobile Verification Toolkit")
|
||||
rich_print("\t\thttps://mvt.re")
|
||||
rich_print(f"\t\tVersion: {MVT_VERSION}")
|
||||
rich_print("\t[bold]MVT - Mobile Verification Toolkit[/bold]\n")
|
||||
rich_print("\thttps://mvt.re")
|
||||
rich_print(f"\tVersion: {MVT_VERSION}\n")
|
||||
|
||||
check_updates(disable_version_check, disable_indicator_check)
|
||||
|
||||
|
||||
@@ -179,21 +179,36 @@ def _module_key(module_class: type[MVTModule]) -> tuple[str, str]:
|
||||
return (source, module_class.__qualname__)
|
||||
|
||||
|
||||
def distribution_direct_url(dist: importlib.metadata.Distribution) -> Optional[dict]:
|
||||
"""Return the PEP 610 direct URL metadata of a distribution, if recorded.
|
||||
|
||||
Packages installed from an index have no direct URL metadata, while
|
||||
packages installed directly from a repository or from a local folder
|
||||
record where they were installed from in ``direct_url.json``.
|
||||
"""
|
||||
try:
|
||||
direct_url_text = dist.read_text("direct_url.json")
|
||||
if not direct_url_text:
|
||||
return None
|
||||
direct_url = json.loads(direct_url_text)
|
||||
return direct_url if isinstance(direct_url, dict) else None
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def _distribution_commit(dist: importlib.metadata.Distribution) -> Optional[str]:
|
||||
"""Return the VCS commit a distribution was installed from, if recorded.
|
||||
|
||||
Packages installed directly from a repository (``pip install git+...``)
|
||||
record the commit in ``direct_url.json`` (PEP 610).
|
||||
"""
|
||||
try:
|
||||
direct_url_text = dist.read_text("direct_url.json")
|
||||
if not direct_url_text:
|
||||
return None
|
||||
commit = json.loads(direct_url_text).get("vcs_info", {}).get("commit_id")
|
||||
return commit if isinstance(commit, str) else None
|
||||
except Exception:
|
||||
vcs_info = (distribution_direct_url(dist) or {}).get("vcs_info")
|
||||
if not isinstance(vcs_info, dict):
|
||||
return None
|
||||
|
||||
commit = vcs_info.get("commit_id")
|
||||
return commit if isinstance(commit, str) else None
|
||||
|
||||
|
||||
def _entry_point_origin(entry_point: importlib.metadata.EntryPoint) -> ModuleOrigin:
|
||||
name = entry_point.name
|
||||
|
||||
+416
-3
@@ -3,8 +3,13 @@
|
||||
# Use of this software is governed by the MVT License 1.1 that can be found at
|
||||
# https://license.mvt.re/1.1/
|
||||
|
||||
import importlib.metadata
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import shlex
|
||||
import subprocess
|
||||
from datetime import datetime
|
||||
from typing import Optional, Tuple
|
||||
|
||||
@@ -12,14 +17,35 @@ import requests
|
||||
import yaml
|
||||
from packaging import version
|
||||
|
||||
from .cli_plugins import (
|
||||
ANDROID_CLI_PLUGIN_GROUP,
|
||||
IOS_CLI_PLUGIN_GROUP,
|
||||
NEUTRAL_CLI_PLUGIN_GROUP,
|
||||
)
|
||||
from .config import settings
|
||||
from .indicators import MVT_DATA_FOLDER, MVT_INDICATORS_FOLDER
|
||||
from .module_loader import MODULES_ENTRY_POINT_GROUP, distribution_direct_url
|
||||
from .version import MVT_VERSION
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
# In hours.
|
||||
INDICATORS_CHECK_FREQUENCY = 12
|
||||
PLUGINS_CHECK_FREQUENCY = 12
|
||||
|
||||
# The entry-point groups a package can use to extend MVT.
|
||||
PLUGIN_ENTRY_POINT_GROUPS = (
|
||||
MODULES_ENTRY_POINT_GROUP,
|
||||
IOS_CLI_PLUGIN_GROUP,
|
||||
ANDROID_CLI_PLUGIN_GROUP,
|
||||
NEUTRAL_CLI_PLUGIN_GROUP,
|
||||
)
|
||||
SHORT_COMMIT_LENGTH = 8
|
||||
# The keys every cached finding has to carry to be printed.
|
||||
FINDING_KEYS = ("name", "installed", "latest", "upgrade_command")
|
||||
# Options which stop ssh from waiting for an answer nobody is there to give.
|
||||
_SSH_BATCH_OPTIONS = ("-o", "BatchMode=yes", "-o", "ConnectTimeout=10")
|
||||
_COMMIT_PATTERN = re.compile(r"\A[0-9a-f]{7,40}\Z")
|
||||
|
||||
|
||||
class MVTUpdates:
|
||||
@@ -38,6 +64,395 @@ class MVTUpdates:
|
||||
return ""
|
||||
|
||||
|
||||
def installed_plugin_distributions() -> list[importlib.metadata.Distribution]:
|
||||
"""Return the installed distributions which extend MVT.
|
||||
|
||||
A plugin package is any distribution registering at least one entry point
|
||||
in the module or CLI command groups. Distributions are returned once each,
|
||||
sorted by name. MVT itself is not a plugin and is never returned.
|
||||
"""
|
||||
distributions: dict[str, importlib.metadata.Distribution] = {}
|
||||
|
||||
for group in PLUGIN_ENTRY_POINT_GROUPS:
|
||||
try:
|
||||
entry_points = importlib.metadata.entry_points(group=group)
|
||||
except Exception as exc:
|
||||
log.warning(
|
||||
"Unable to discover installed plugin packages in entry-point "
|
||||
"group %s: %s",
|
||||
group,
|
||||
exc,
|
||||
)
|
||||
continue
|
||||
|
||||
for entry_point in entry_points:
|
||||
# Manually constructed entry points have no associated distribution.
|
||||
dist = getattr(entry_point, "dist", None)
|
||||
if dist is None:
|
||||
continue
|
||||
try:
|
||||
name = dist.name
|
||||
except Exception:
|
||||
continue
|
||||
if not name or name == "mvt":
|
||||
continue
|
||||
distributions.setdefault(name, dist)
|
||||
|
||||
return [distributions[name] for name in sorted(distributions)]
|
||||
|
||||
|
||||
def _is_usable_finding(finding: object) -> bool:
|
||||
"""Check that a cached finding carries everything needed to print it."""
|
||||
if not isinstance(finding, dict):
|
||||
return False
|
||||
|
||||
return all(
|
||||
isinstance(finding.get(key), str) and finding.get(key) for key in FINDING_KEYS
|
||||
)
|
||||
|
||||
|
||||
def _installed_revision(
|
||||
dist: importlib.metadata.Distribution, origin: object
|
||||
) -> Optional[str]:
|
||||
"""Return what a plugin's installed revision is right now.
|
||||
|
||||
This is the value a finding recorded as installed when it was made, so a
|
||||
finding can be compared against the current state of the installation
|
||||
without looking anything up remotely.
|
||||
"""
|
||||
try:
|
||||
if origin != "git":
|
||||
return dist.version
|
||||
|
||||
vcs_info = (distribution_direct_url(dist) or {}).get("vcs_info")
|
||||
if not isinstance(vcs_info, dict):
|
||||
return None
|
||||
|
||||
commit = vcs_info.get("commit_id") or ""
|
||||
return commit[:SHORT_COMMIT_LENGTH] or None
|
||||
except Exception as e:
|
||||
log.debug("Failed to read the installed revision of a plugin: %s", e)
|
||||
return None
|
||||
|
||||
|
||||
def _batch_mode_ssh_command(ssh_command: str) -> str:
|
||||
"""Return an ssh command line which cannot stop to ask a question.
|
||||
|
||||
ssh keeps the first value it is given for a keyword, so the batch mode
|
||||
options are inserted right after the ssh program, ahead of whatever the
|
||||
analyst configured. Their remaining options, such as which key to use,
|
||||
still apply.
|
||||
"""
|
||||
try:
|
||||
arguments = shlex.split(ssh_command.strip())
|
||||
except ValueError:
|
||||
arguments = []
|
||||
|
||||
if not arguments:
|
||||
arguments = ["ssh"]
|
||||
|
||||
return shlex.join([arguments[0], *_SSH_BATCH_OPTIONS, *arguments[1:]])
|
||||
|
||||
|
||||
def _revision_pins_commit(revision: str, commit: str) -> bool:
|
||||
"""Check whether a requested revision pins the installed commit."""
|
||||
candidate = revision.lower()
|
||||
if not _COMMIT_PATTERN.match(candidate):
|
||||
return False
|
||||
|
||||
return commit.lower().startswith(candidate)
|
||||
|
||||
|
||||
class PluginUpdates:
|
||||
"""Check for updates to the installed MVT plugin packages.
|
||||
|
||||
MVT never installs or upgrades a plugin package itself. It only reports
|
||||
the command which upgrades a plugin, leaving the analyst to decide when to
|
||||
run it.
|
||||
"""
|
||||
|
||||
@property
|
||||
def latest_check_path(self) -> str:
|
||||
return os.path.join(MVT_DATA_FOLDER, "latest_plugins_check")
|
||||
|
||||
@property
|
||||
def findings_path(self) -> str:
|
||||
return os.path.join(MVT_DATA_FOLDER, "plugin_updates.json")
|
||||
|
||||
def _create_data_folder(self) -> None:
|
||||
if not os.path.exists(MVT_DATA_FOLDER):
|
||||
os.makedirs(MVT_DATA_FOLDER)
|
||||
|
||||
def get_latest_check(self) -> int:
|
||||
if not os.path.exists(self.latest_check_path):
|
||||
return 0
|
||||
|
||||
# A corrupt or truncated timestamp only means the next check happens
|
||||
# sooner. It must never stop MVT from running.
|
||||
try:
|
||||
with open(self.latest_check_path, "r", encoding="utf-8") as handle:
|
||||
data = handle.read().strip()
|
||||
if data:
|
||||
return int(data)
|
||||
except (OSError, ValueError) as e:
|
||||
log.debug("Failed to read the time of the latest plugin check: %s", e)
|
||||
|
||||
return 0
|
||||
|
||||
def set_latest_check(self) -> None:
|
||||
self._create_data_folder()
|
||||
timestamp = int(datetime.now().timestamp())
|
||||
with open(self.latest_check_path, "w", encoding="utf-8") as handle:
|
||||
handle.write(str(timestamp))
|
||||
|
||||
def get_findings(self) -> list[dict]:
|
||||
"""
|
||||
Return the findings of the latest check, without checking again.
|
||||
Returns an empty list if no check was ever performed.
|
||||
"""
|
||||
if not os.path.exists(self.findings_path):
|
||||
return []
|
||||
|
||||
try:
|
||||
with open(self.findings_path, "r", encoding="utf-8") as handle:
|
||||
findings = json.load(handle)
|
||||
except Exception as e:
|
||||
log.debug("Failed to read the cached plugin updates: %s", e)
|
||||
return []
|
||||
|
||||
if not isinstance(findings, list):
|
||||
return []
|
||||
|
||||
# Anything which does not look like a finding is dropped rather than
|
||||
# trusted: the cache is only a convenience.
|
||||
return [finding for finding in findings if _is_usable_finding(finding)]
|
||||
|
||||
def current_findings(
|
||||
self, distributions: Optional[list[importlib.metadata.Distribution]] = None
|
||||
) -> list[dict]:
|
||||
"""
|
||||
Return the cached findings which still apply to what is installed.
|
||||
Findings about a plugin which was upgraded or removed since the latest
|
||||
check are dropped, so an update is never reported twice.
|
||||
"""
|
||||
if distributions is None:
|
||||
distributions = installed_plugin_distributions()
|
||||
|
||||
installed = {}
|
||||
for dist in distributions:
|
||||
try:
|
||||
installed[dist.name] = dist
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
current = []
|
||||
for finding in self.get_findings():
|
||||
plugin = installed.get(finding["name"])
|
||||
if plugin is None:
|
||||
continue
|
||||
if (
|
||||
_installed_revision(plugin, finding.get("origin"))
|
||||
!= finding["installed"]
|
||||
):
|
||||
continue
|
||||
current.append(finding)
|
||||
|
||||
return current
|
||||
|
||||
def set_findings(self, findings: list[dict]) -> None:
|
||||
self._create_data_folder()
|
||||
with open(self.findings_path, "w", encoding="utf-8") as handle:
|
||||
json.dump(findings, handle)
|
||||
|
||||
def should_check(self) -> Tuple[bool, int]:
|
||||
"""
|
||||
Compare time of the latest plugins check with current time.
|
||||
Returns bool and number of hours since the last check.
|
||||
"""
|
||||
now = datetime.now()
|
||||
latest_check_ts = self.get_latest_check()
|
||||
latest_check_dt = datetime.fromtimestamp(latest_check_ts)
|
||||
|
||||
diff = now - latest_check_dt
|
||||
diff_hours = divmod(diff.total_seconds(), 3600)[0]
|
||||
|
||||
if diff_hours >= PLUGINS_CHECK_FREQUENCY:
|
||||
return True, 0
|
||||
|
||||
return False, int(PLUGINS_CHECK_FREQUENCY - diff_hours)
|
||||
|
||||
def _check_index_plugin(self, name: str, installed: str) -> Optional[dict]:
|
||||
"""Check a plugin installed from a package index for a newer release."""
|
||||
url = f"https://pypi.org/pypi/{name}/json"
|
||||
try:
|
||||
res = requests.get(url, timeout=settings.NETWORK_TIMEOUT)
|
||||
except requests.exceptions.RequestException as e:
|
||||
log.debug("Failed to check for updates to plugin %s: %s", name, e)
|
||||
return None
|
||||
|
||||
# Plugins which were never published to a public index are expected,
|
||||
# and there is nothing to compare their version against.
|
||||
if res.status_code == 404:
|
||||
return None
|
||||
|
||||
if res.status_code != 200:
|
||||
log.debug(
|
||||
"Failed to check for updates to plugin %s (error %d)",
|
||||
name,
|
||||
res.status_code,
|
||||
)
|
||||
return None
|
||||
|
||||
try:
|
||||
latest = res.json().get("info", {}).get("version", "")
|
||||
if not latest or version.parse(latest) <= version.parse(installed):
|
||||
return None
|
||||
except Exception as e:
|
||||
log.debug("Failed to compare the versions of plugin %s: %s", name, e)
|
||||
return None
|
||||
|
||||
return {
|
||||
"name": name,
|
||||
"installed": installed,
|
||||
"latest": latest,
|
||||
"origin": "pypi",
|
||||
# The name comes from package metadata, so the command MVT
|
||||
# suggests is quoted rather than assumed to be shell-safe.
|
||||
"upgrade_command": f"pip install -U {shlex.quote(name)}",
|
||||
}
|
||||
|
||||
def _git_ls_remote(self, url: str, revision: str) -> list[Tuple[str, str]]:
|
||||
"""Return the remote references matching a revision, if git allows it."""
|
||||
# Neither value is trusted: they are read from the metadata of an
|
||||
# installed package and must not turn into git options.
|
||||
if url.startswith("-") or revision.startswith("-"):
|
||||
log.debug("Skipping the update check for the invalid repository %s", url)
|
||||
return []
|
||||
|
||||
environment = dict(os.environ)
|
||||
# Never prompt the analyst for repository credentials. git handles its
|
||||
# own prompts, while ssh reads the terminal directly and only batch
|
||||
# mode makes it fail instead of asking.
|
||||
environment["GIT_TERMINAL_PROMPT"] = "0"
|
||||
environment["GIT_SSH_COMMAND"] = _batch_mode_ssh_command(
|
||||
environment.get("GIT_SSH_COMMAND", "")
|
||||
)
|
||||
|
||||
try:
|
||||
process = subprocess.run(
|
||||
["git", "ls-remote", url, revision],
|
||||
capture_output=True,
|
||||
stdin=subprocess.DEVNULL,
|
||||
text=True,
|
||||
env=environment,
|
||||
timeout=settings.NETWORK_TIMEOUT,
|
||||
check=False,
|
||||
)
|
||||
except FileNotFoundError:
|
||||
log.debug("Could not find git, skipping the update check for %s", url)
|
||||
return []
|
||||
except (subprocess.SubprocessError, OSError) as e:
|
||||
log.debug("Failed to query the repository %s: %s", url, e)
|
||||
return []
|
||||
|
||||
if process.returncode != 0:
|
||||
log.debug(
|
||||
"Failed to query the repository %s (error %d): %s",
|
||||
url,
|
||||
process.returncode,
|
||||
(process.stderr or "").strip(),
|
||||
)
|
||||
return []
|
||||
|
||||
references = []
|
||||
for line in (process.stdout or "").splitlines():
|
||||
commit, _, reference = line.partition("\t")
|
||||
if commit.strip() and reference.strip():
|
||||
references.append((commit.strip(), reference.strip()))
|
||||
|
||||
return references
|
||||
|
||||
def _check_repository_plugin(
|
||||
self, name: str, direct_url: dict, vcs_info: dict
|
||||
) -> Optional[dict]:
|
||||
"""Check a plugin installed from a repository for a newer commit."""
|
||||
url = direct_url.get("url") or ""
|
||||
installed = vcs_info.get("commit_id") or ""
|
||||
revision = vcs_info.get("requested_revision") or ""
|
||||
if not url or not installed:
|
||||
return None
|
||||
|
||||
# A plugin installed from a commit is pinned and never goes out of
|
||||
# date, no matter what the branch it came from does next.
|
||||
if revision and _revision_pins_commit(revision, installed):
|
||||
return None
|
||||
|
||||
latest = ""
|
||||
wanted_reference = f"refs/heads/{revision}" if revision else "HEAD"
|
||||
for commit, reference in self._git_ls_remote(url, revision or "HEAD"):
|
||||
# Tags are pinned installs too.
|
||||
if reference.startswith("refs/tags/"):
|
||||
return None
|
||||
if reference == wanted_reference:
|
||||
latest = commit
|
||||
|
||||
if not latest or latest == installed:
|
||||
return None
|
||||
|
||||
requirement = f"{name} @ git+{url}"
|
||||
if revision:
|
||||
requirement += f"@{revision}"
|
||||
|
||||
return {
|
||||
"name": name,
|
||||
"installed": installed[:SHORT_COMMIT_LENGTH],
|
||||
"latest": latest[:SHORT_COMMIT_LENGTH],
|
||||
"origin": "git",
|
||||
# A repository URL and a branch name can both hold characters a
|
||||
# shell would act on, so the requirement is quoted for the shell
|
||||
# the analyst is going to paste the command into.
|
||||
"upgrade_command": f"pip install -U {shlex.quote(requirement)}",
|
||||
}
|
||||
|
||||
def _check_distribution(
|
||||
self, dist: importlib.metadata.Distribution
|
||||
) -> Optional[dict]:
|
||||
try:
|
||||
name = dist.name
|
||||
installed = dist.version
|
||||
except Exception as e:
|
||||
log.debug("Failed to read the metadata of an installed plugin: %s", e)
|
||||
return None
|
||||
|
||||
direct_url = distribution_direct_url(dist)
|
||||
if direct_url is None:
|
||||
return self._check_index_plugin(name, installed)
|
||||
|
||||
vcs_info = direct_url.get("vcs_info")
|
||||
if isinstance(vcs_info, dict):
|
||||
return self._check_repository_plugin(name, direct_url, vcs_info)
|
||||
|
||||
# Plugins installed from a local folder, including editable installs,
|
||||
# are maintained by the analyst and have nothing to check against.
|
||||
return None
|
||||
|
||||
def check(self) -> list[dict]:
|
||||
"""
|
||||
Check every installed plugin package for an available update.
|
||||
Returns one entry per plugin which can be upgraded.
|
||||
"""
|
||||
findings = []
|
||||
for dist in installed_plugin_distributions():
|
||||
finding = self._check_distribution(dist)
|
||||
if finding:
|
||||
findings.append(finding)
|
||||
|
||||
self.set_findings(findings)
|
||||
self.set_latest_check()
|
||||
|
||||
return findings
|
||||
|
||||
|
||||
class IndicatorsUpdates:
|
||||
def __init__(self) -> None:
|
||||
self.github_raw_url = "https://raw.githubusercontent.com/{}/{}/{}/{}"
|
||||
@@ -180,9 +595,7 @@ class IndicatorsUpdates:
|
||||
def _get_remote_file_latest_commit(
|
||||
self, owner: str, repo: str, branch: str, path: str
|
||||
) -> int:
|
||||
file_commit_url = (
|
||||
f"https://api.github.com/repos/{owner}/{repo}/commits?path={path}&sha={branch}"
|
||||
)
|
||||
file_commit_url = f"https://api.github.com/repos/{owner}/{repo}/commits?path={path}&sha={branch}"
|
||||
try:
|
||||
res = requests.get(file_commit_url, timeout=5)
|
||||
except requests.exceptions.RequestException as e:
|
||||
|
||||
+5
-47
@@ -15,12 +15,6 @@ from mvt.common.cli_plugins import (
|
||||
register_cli_plugins,
|
||||
)
|
||||
from mvt.common.cmd_check_iocs import CmdCheckIOCS
|
||||
from mvt.common.completion import (
|
||||
SUPPORTED_SHELLS,
|
||||
completion_instructions,
|
||||
generate_completion_script,
|
||||
install_completion_script,
|
||||
)
|
||||
from mvt.common.logo import logo
|
||||
from mvt.common.options import MutuallyExclusiveOption
|
||||
from mvt.common.updates import IndicatorsUpdates
|
||||
@@ -51,7 +45,6 @@ from mvt.common.help import (
|
||||
HELP_MSG_CHECK_SYSDIAGNOSE,
|
||||
HELP_MSG_DISABLE_UPDATE_CHECK,
|
||||
HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK,
|
||||
HELP_MSG_COMPLETION,
|
||||
)
|
||||
from mvt.common.module_loader import CustomModuleLoadError, load_custom_modules
|
||||
from mvt.common.password import prompt_password
|
||||
@@ -106,55 +99,20 @@ def cli(ctx, disable_update_check, disable_indicator_update_check):
|
||||
ctx.ensure_object(dict)
|
||||
ctx.obj["disable_version_check"] = disable_update_check
|
||||
ctx.obj["disable_indicator_check"] = disable_indicator_update_check
|
||||
if ctx.invoked_subcommand != "completion":
|
||||
logo(
|
||||
disable_version_check=disable_update_check,
|
||||
disable_indicator_check=disable_indicator_update_check,
|
||||
)
|
||||
logo(
|
||||
disable_version_check=disable_update_check,
|
||||
disable_indicator_check=disable_indicator_update_check,
|
||||
)
|
||||
|
||||
|
||||
# ==============================================================================
|
||||
# Command: version
|
||||
# ==============================================================================
|
||||
@cli.command("version", help=HELP_MSG_VERSION)
|
||||
@cli.command("version", context_settings=CONTEXT_SETTINGS, help=HELP_MSG_VERSION)
|
||||
def version():
|
||||
return
|
||||
|
||||
|
||||
# ==============================================================================
|
||||
# Command: completion
|
||||
# ==============================================================================
|
||||
@cli.command("completion", context_settings=CONTEXT_SETTINGS, help=HELP_MSG_COMPLETION)
|
||||
@click.argument("shell", required=False, type=click.Choice(SUPPORTED_SHELLS))
|
||||
@click.option(
|
||||
"--install",
|
||||
is_flag=True,
|
||||
help="Write completion files and update shell configuration.",
|
||||
)
|
||||
@click.pass_context
|
||||
def completion(ctx, shell, install):
|
||||
program_name = "mvt-ios"
|
||||
|
||||
if shell is None:
|
||||
if install:
|
||||
raise click.UsageError("A shell is required when using --install.")
|
||||
click.echo(completion_instructions(program_name))
|
||||
return
|
||||
|
||||
root_cli = ctx.find_root().command
|
||||
|
||||
if install:
|
||||
script_path = install_completion_script(root_cli, program_name, shell)
|
||||
click.echo(f"Installed {shell} completion to {script_path}")
|
||||
if shell in ("bash", "zsh"):
|
||||
click.echo(f"Updated ~/.{shell}rc")
|
||||
else:
|
||||
click.echo("Fish loads completion files automatically.")
|
||||
return
|
||||
|
||||
click.echo(generate_completion_script(root_cli, program_name, shell))
|
||||
|
||||
|
||||
# ==============================================================================
|
||||
# Command: decrypt-backup
|
||||
# ==============================================================================
|
||||
|
||||
@@ -3,9 +3,12 @@ from types import SimpleNamespace
|
||||
import click
|
||||
from click.testing import CliRunner
|
||||
|
||||
from mvt.cli import cli as mvt_cli
|
||||
from mvt.common.cli_plugins import (
|
||||
ANDROID_CLI_PLUGIN_GROUP,
|
||||
IOS_CLI_PLUGIN_GROUP,
|
||||
MVT_CUSTOM_COMMANDS_ENV,
|
||||
NEUTRAL_CLI_PLUGIN_GROUP,
|
||||
BrokenPluginCommand,
|
||||
load_cli_commands_option,
|
||||
register_cli_commands_from_path,
|
||||
@@ -14,6 +17,9 @@ from mvt.common.cli_plugins import (
|
||||
)
|
||||
|
||||
|
||||
# Keep the banner of the mvt group callback from checking for updates online.
|
||||
OFFLINE = ["--disable-update-check", "--disable-indicator-update-check"]
|
||||
|
||||
COMMAND_TEMPLATE = """
|
||||
import click
|
||||
|
||||
@@ -345,7 +351,11 @@ def test_platform_entry_point_groups_and_environment_paths_are_separate(
|
||||
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)]
|
||||
if group == ANDROID_CLI_PLUGIN_GROUP:
|
||||
return [
|
||||
_entry_point("android-package", "android_plugin:cli", android_package)
|
||||
]
|
||||
return []
|
||||
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.cli_plugins.importlib.metadata.entry_points",
|
||||
@@ -369,3 +379,187 @@ def test_platform_entry_point_groups_and_environment_paths_are_separate(
|
||||
|
||||
assert set(ios_group.commands) == {"ios-file", "ios-package"}
|
||||
assert set(android_group.commands) == {"android-file", "android-package"}
|
||||
|
||||
|
||||
def test_neutral_entry_point_group_is_not_registered_on_the_platform_clis(
|
||||
monkeypatch,
|
||||
):
|
||||
@click.command()
|
||||
def neutral_package():
|
||||
pass
|
||||
|
||||
def entry_points(*, group):
|
||||
if group == NEUTRAL_CLI_PLUGIN_GROUP:
|
||||
return [
|
||||
_entry_point("neutral-package", "neutral_plugin:cli", neutral_package)
|
||||
]
|
||||
return []
|
||||
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.cli_plugins.importlib.metadata.entry_points",
|
||||
entry_points,
|
||||
)
|
||||
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 not ios_group.commands
|
||||
assert not android_group.commands
|
||||
|
||||
|
||||
def test_environment_command_wins_collision_with_installed_command(
|
||||
tmp_path, monkeypatch, caplog
|
||||
):
|
||||
command_path = _write_command(
|
||||
tmp_path / "duplicate.py",
|
||||
"duplicate",
|
||||
message="environment command ran",
|
||||
)
|
||||
|
||||
@click.command()
|
||||
def installed_command():
|
||||
pass
|
||||
|
||||
def entry_points(*, group):
|
||||
if group == IOS_CLI_PLUGIN_GROUP:
|
||||
return [
|
||||
_entry_point(
|
||||
"duplicate",
|
||||
"ios_plugin:cli",
|
||||
installed_command,
|
||||
distribution="ios-plugin",
|
||||
)
|
||||
]
|
||||
return []
|
||||
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.cli_plugins.importlib.metadata.entry_points",
|
||||
entry_points,
|
||||
)
|
||||
monkeypatch.setenv("TEST_IOS_COMMANDS", str(command_path))
|
||||
group = click.Group()
|
||||
|
||||
register_cli_plugins(
|
||||
group,
|
||||
entry_point_group=IOS_CLI_PLUGIN_GROUP,
|
||||
environment_variable="TEST_IOS_COMMANDS",
|
||||
)
|
||||
|
||||
assert group.commands["duplicate"] is not installed_command
|
||||
result = CliRunner().invoke(group, ["duplicate"])
|
||||
assert result.exit_code == 0
|
||||
assert "environment command ran" in result.output
|
||||
assert "the command name is already registered" in caplog.text
|
||||
assert "ios-plugin 1.0 (ios_plugin:cli)" in caplog.text
|
||||
|
||||
|
||||
def test_the_mvt_cli_gets_the_neutral_commands_and_no_platform_command(
|
||||
monkeypatch, restore_cli_commands
|
||||
):
|
||||
@click.command()
|
||||
def shared_package():
|
||||
click.echo("shared command ran")
|
||||
|
||||
@click.command()
|
||||
def ios_package():
|
||||
pass
|
||||
|
||||
def entry_points(*, group):
|
||||
if group == NEUTRAL_CLI_PLUGIN_GROUP:
|
||||
return [_entry_point("shared-package", "shared_plugin:cli", shared_package)]
|
||||
if group == IOS_CLI_PLUGIN_GROUP:
|
||||
return [_entry_point("ios-package", "ios_plugin:cli", ios_package)]
|
||||
return []
|
||||
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.cli_plugins.importlib.metadata.entry_points",
|
||||
entry_points,
|
||||
)
|
||||
|
||||
register_cli_plugins(
|
||||
mvt_cli,
|
||||
entry_point_group=NEUTRAL_CLI_PLUGIN_GROUP,
|
||||
environment_variable=MVT_CUSTOM_COMMANDS_ENV,
|
||||
)
|
||||
|
||||
assert "ios-package" not in mvt_cli.commands
|
||||
result = CliRunner().invoke(mvt_cli, [*OFFLINE, "shared-package"])
|
||||
assert result.exit_code == 0
|
||||
assert "shared command ran" in result.output
|
||||
|
||||
|
||||
def test_builtin_mvt_command_wins_collision_with_neutral_command(
|
||||
monkeypatch, caplog, restore_cli_commands
|
||||
):
|
||||
@click.command()
|
||||
def neutral_version():
|
||||
pass
|
||||
|
||||
def entry_points(*, group):
|
||||
if group == NEUTRAL_CLI_PLUGIN_GROUP:
|
||||
return [
|
||||
_entry_point(
|
||||
"version",
|
||||
"neutral_plugin:cli",
|
||||
neutral_version,
|
||||
distribution="neutral-plugin",
|
||||
)
|
||||
]
|
||||
return []
|
||||
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.cli_plugins.importlib.metadata.entry_points",
|
||||
entry_points,
|
||||
)
|
||||
builtin_version = mvt_cli.commands["version"]
|
||||
|
||||
register_cli_plugins(
|
||||
mvt_cli,
|
||||
entry_point_group=NEUTRAL_CLI_PLUGIN_GROUP,
|
||||
environment_variable=MVT_CUSTOM_COMMANDS_ENV,
|
||||
)
|
||||
|
||||
assert mvt_cli.commands["version"] is builtin_version
|
||||
assert "the command name is already registered" in caplog.text
|
||||
assert "neutral-plugin 1.0 (neutral_plugin:cli)" in caplog.text
|
||||
|
||||
|
||||
def test_broken_neutral_plugin_does_not_break_the_mvt_cli(
|
||||
monkeypatch, restore_cli_commands
|
||||
):
|
||||
def entry_points(*, group):
|
||||
if group == NEUTRAL_CLI_PLUGIN_GROUP:
|
||||
return [
|
||||
_entry_point(
|
||||
"broken",
|
||||
"broken_plugin:cli",
|
||||
exception=RuntimeError("missing dependency"),
|
||||
distribution="broken-plugin",
|
||||
)
|
||||
]
|
||||
return []
|
||||
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.cli_plugins.importlib.metadata.entry_points",
|
||||
entry_points,
|
||||
)
|
||||
|
||||
register_cli_plugins(
|
||||
mvt_cli,
|
||||
entry_point_group=NEUTRAL_CLI_PLUGIN_GROUP,
|
||||
environment_variable=MVT_CUSTOM_COMMANDS_ENV,
|
||||
)
|
||||
|
||||
assert isinstance(mvt_cli.commands["broken"], BrokenPluginCommand)
|
||||
result = CliRunner().invoke(mvt_cli, [*OFFLINE, "version"])
|
||||
assert result.exit_code == 0
|
||||
@@ -0,0 +1,236 @@
|
||||
import json
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
from click.testing import CliRunner
|
||||
|
||||
from mvt.android.cli import cli as android_cli
|
||||
from mvt.cli import cli as mvt_cli
|
||||
from mvt.common.cli_plugins import (
|
||||
ANDROID_CLI_PLUGIN_GROUP,
|
||||
IOS_CLI_PLUGIN_GROUP,
|
||||
NEUTRAL_CLI_PLUGIN_GROUP,
|
||||
)
|
||||
from mvt.common.cmd_plugins import plugins
|
||||
from mvt.common.module import MVTModule
|
||||
from mvt.common.module_loader import MODULES_ENTRY_POINT_GROUP
|
||||
from mvt.common.updates import PluginUpdates
|
||||
from mvt.ios.cli import cli as ios_cli
|
||||
|
||||
|
||||
class ExampleModule(MVTModule):
|
||||
pass
|
||||
|
||||
|
||||
class AnotherModule(MVTModule):
|
||||
pass
|
||||
|
||||
|
||||
class FakeDistribution:
|
||||
def __init__(self, name, version="1.0.0", direct_url=None):
|
||||
self.name = name
|
||||
self.version = version
|
||||
self.direct_url = direct_url
|
||||
|
||||
def read_text(self, file_name):
|
||||
if file_name == "direct_url.json" and self.direct_url is not None:
|
||||
return json.dumps(self.direct_url)
|
||||
return None
|
||||
|
||||
|
||||
def _entry_point(name, distribution, modules=None, exception=None):
|
||||
def load():
|
||||
if exception is not None:
|
||||
raise exception
|
||||
return modules
|
||||
|
||||
return SimpleNamespace(
|
||||
name=name, value="example_plugin:modules", dist=distribution, load=load
|
||||
)
|
||||
|
||||
|
||||
def _run(command, arguments):
|
||||
# Keep rich from wrapping the table while its content is being asserted.
|
||||
return CliRunner().invoke(command, arguments, env={"COLUMNS": "200"})
|
||||
|
||||
|
||||
def _table_rows(output):
|
||||
"""Return the content of the table rows, without the header and the box."""
|
||||
return [
|
||||
[cell.strip() for cell in line.strip().strip("│").split("│")]
|
||||
for line in output.splitlines()
|
||||
if "│" in line
|
||||
]
|
||||
|
||||
|
||||
def _table_header(output):
|
||||
for line in output.splitlines():
|
||||
if "┃" in line:
|
||||
return [cell.strip() for cell in line.strip().strip("┃").split("┃")]
|
||||
return []
|
||||
|
||||
|
||||
def _install(monkeypatch, distributions, entry_points):
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.cmd_plugins.installed_plugin_distributions",
|
||||
lambda: distributions,
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.cmd_plugins.importlib.metadata.entry_points",
|
||||
lambda *, group: entry_points.get(group, []),
|
||||
)
|
||||
|
||||
|
||||
def test_plugins_is_a_builtin_command_of_the_mvt_cli_only():
|
||||
assert mvt_cli.commands["plugins"] is plugins
|
||||
assert "plugins" not in ios_cli.commands
|
||||
assert "plugins" not in android_cli.commands
|
||||
|
||||
|
||||
def test_list_shows_what_every_plugin_contributes(monkeypatch):
|
||||
index_plugin = FakeDistribution("example-plugin", version="1.2.0")
|
||||
repository_plugin = FakeDistribution(
|
||||
"repository-plugin",
|
||||
version="0.1.0",
|
||||
direct_url={
|
||||
"url": "https://example.org/plugin.git",
|
||||
"vcs_info": {"vcs": "git", "commit_id": "b" * 40},
|
||||
},
|
||||
)
|
||||
local_plugin = FakeDistribution(
|
||||
"local-plugin",
|
||||
direct_url={"url": "file:///plugins", "dir_info": {"editable": True}},
|
||||
)
|
||||
_install(
|
||||
monkeypatch,
|
||||
[index_plugin, local_plugin, repository_plugin],
|
||||
{
|
||||
MODULES_ENTRY_POINT_GROUP: [
|
||||
_entry_point(
|
||||
"example", index_plugin, modules=[ExampleModule, AnotherModule]
|
||||
),
|
||||
_entry_point("local", local_plugin, modules=lambda: [ExampleModule]),
|
||||
],
|
||||
IOS_CLI_PLUGIN_GROUP: [_entry_point("summarize", repository_plugin)],
|
||||
ANDROID_CLI_PLUGIN_GROUP: [_entry_point("triage", local_plugin)],
|
||||
NEUTRAL_CLI_PLUGIN_GROUP: [_entry_point("report", repository_plugin)],
|
||||
},
|
||||
)
|
||||
|
||||
result = _run(plugins, ["list"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
# Plugins are listed by name, with the modules and the commands each of
|
||||
# them contributes.
|
||||
assert _table_header(result.output) == [
|
||||
"Name",
|
||||
"Version",
|
||||
"Origin",
|
||||
"Modules",
|
||||
"Commands",
|
||||
]
|
||||
assert _table_rows(result.output) == [
|
||||
["example-plugin", "1.2.0", "pypi", "2", "-"],
|
||||
["local-plugin", "1.0.0", "local", "1", "triage"],
|
||||
["repository-plugin", "0.1.0", "git+bbbbbbbb", "0", "report, summarize"],
|
||||
]
|
||||
|
||||
|
||||
def test_list_reports_a_broken_module_entry_point(monkeypatch):
|
||||
plugin = FakeDistribution("broken-plugin")
|
||||
_install(
|
||||
monkeypatch,
|
||||
[plugin],
|
||||
{
|
||||
MODULES_ENTRY_POINT_GROUP: [
|
||||
_entry_point(
|
||||
"broken", plugin, exception=ImportError("missing dependency")
|
||||
)
|
||||
]
|
||||
},
|
||||
)
|
||||
|
||||
result = _run(plugins, ["list"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert _table_rows(result.output) == [
|
||||
["broken-plugin", "1.0.0", "pypi", "error", "-"]
|
||||
]
|
||||
|
||||
|
||||
def test_list_without_plugins(monkeypatch):
|
||||
_install(monkeypatch, [], {})
|
||||
|
||||
result = _run(plugins, ["list"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert result.output == "No MVT plugins are installed.\n"
|
||||
|
||||
|
||||
def test_check_updates_prints_the_findings_and_ignores_the_throttle(monkeypatch):
|
||||
findings = [
|
||||
{
|
||||
"name": "example-plugin",
|
||||
"installed": "1.0.0",
|
||||
"latest": "1.2.0",
|
||||
"origin": "pypi",
|
||||
"upgrade_command": "pip install -U example-plugin",
|
||||
}
|
||||
]
|
||||
_install(monkeypatch, [FakeDistribution("example-plugin")], {})
|
||||
monkeypatch.setattr(PluginUpdates, "check", lambda self: findings)
|
||||
monkeypatch.setattr(
|
||||
PluginUpdates,
|
||||
"should_check",
|
||||
lambda self: pytest.fail("an explicit check must not be throttled"),
|
||||
)
|
||||
|
||||
result = _run(plugins, ["check-updates"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "Plugin updates available:" in result.output
|
||||
assert "example-plugin 1.0.0 → 1.2.0" in result.output
|
||||
assert "Upgrade with: pip install -U example-plugin" in result.output
|
||||
assert "MVT does not install plugin updates." in result.output
|
||||
|
||||
|
||||
def test_check_updates_without_available_updates(monkeypatch):
|
||||
_install(monkeypatch, [FakeDistribution("example-plugin")], {})
|
||||
monkeypatch.setattr(PluginUpdates, "check", lambda self: [])
|
||||
|
||||
result = _run(plugins, ["check-updates"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "All plugins are up to date." in result.output
|
||||
|
||||
|
||||
def test_check_updates_without_plugins(monkeypatch):
|
||||
_install(monkeypatch, [], {})
|
||||
monkeypatch.setattr(
|
||||
PluginUpdates,
|
||||
"check",
|
||||
lambda self: pytest.fail("nothing must be checked without plugins"),
|
||||
)
|
||||
|
||||
result = _run(plugins, ["check-updates"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "No MVT plugins are installed." in result.output
|
||||
|
||||
|
||||
def test_check_updates_without_network_access(monkeypatch):
|
||||
monkeypatch.setattr("mvt.common.cmd_plugins.settings.NETWORK_ACCESS_ALLOWED", False)
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.cmd_plugins.installed_plugin_distributions",
|
||||
lambda: pytest.fail("plugins must not be listed without network access"),
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
PluginUpdates,
|
||||
"check",
|
||||
lambda self: pytest.fail("nothing must be checked without network access"),
|
||||
)
|
||||
|
||||
result = _run(plugins, ["check-updates"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "Network access is disabled" in result.output
|
||||
@@ -0,0 +1,711 @@
|
||||
import json
|
||||
import shlex
|
||||
from datetime import datetime, timedelta
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
from mvt.common import logo
|
||||
from mvt.common.cli_plugins import (
|
||||
ANDROID_CLI_PLUGIN_GROUP,
|
||||
IOS_CLI_PLUGIN_GROUP,
|
||||
NEUTRAL_CLI_PLUGIN_GROUP,
|
||||
)
|
||||
from mvt.common.module_loader import MODULES_ENTRY_POINT_GROUP
|
||||
from mvt.common.updates import (
|
||||
MVTUpdates,
|
||||
PluginUpdates,
|
||||
installed_plugin_distributions,
|
||||
)
|
||||
|
||||
|
||||
REPOSITORY_URL = "https://example.org/plugin.git"
|
||||
INSTALLED_COMMIT = "a" * 40
|
||||
REMOTE_COMMIT = "b" * 40
|
||||
|
||||
|
||||
class FakeDistribution:
|
||||
def __init__(self, name, version="1.0.0", direct_url=None):
|
||||
self.name = name
|
||||
self.version = version
|
||||
self.direct_url = direct_url
|
||||
|
||||
def read_text(self, file_name):
|
||||
if file_name == "direct_url.json" and self.direct_url is not None:
|
||||
return json.dumps(self.direct_url)
|
||||
return None
|
||||
|
||||
|
||||
class FakeResponse:
|
||||
def __init__(self, status_code=200, payload=None):
|
||||
self.status_code = status_code
|
||||
self.payload = payload or {}
|
||||
|
||||
def json(self):
|
||||
return self.payload
|
||||
|
||||
|
||||
def _entry_point(name, distribution, value="plugin:modules"):
|
||||
return SimpleNamespace(name=name, value=value, dist=distribution)
|
||||
|
||||
|
||||
def _git_distribution(requested_revision=None, commit=INSTALLED_COMMIT):
|
||||
vcs_info = {"vcs": "git", "commit_id": commit}
|
||||
if requested_revision:
|
||||
vcs_info["requested_revision"] = requested_revision
|
||||
|
||||
return FakeDistribution(
|
||||
"example-plugin",
|
||||
direct_url={"url": REPOSITORY_URL, "vcs_info": vcs_info},
|
||||
)
|
||||
|
||||
|
||||
def _fake_git(stdout="", returncode=0, exception=None, calls=None):
|
||||
def run(command, **kwargs):
|
||||
if calls is not None:
|
||||
calls.append((command, kwargs))
|
||||
if exception is not None:
|
||||
raise exception
|
||||
return SimpleNamespace(returncode=returncode, stdout=stdout, stderr="")
|
||||
|
||||
return run
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def data_folder(tmp_path, monkeypatch):
|
||||
folder = tmp_path / "mvt-data"
|
||||
monkeypatch.setattr("mvt.common.updates.MVT_DATA_FOLDER", str(folder))
|
||||
return folder
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def one_plugin(monkeypatch):
|
||||
def install(distribution):
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.updates.installed_plugin_distributions",
|
||||
lambda: [distribution],
|
||||
)
|
||||
return distribution
|
||||
|
||||
return install
|
||||
|
||||
|
||||
def test_installed_plugin_distributions_covers_every_plugin_group(monkeypatch):
|
||||
zeta = FakeDistribution("zeta-plugin")
|
||||
alpha = FakeDistribution("alpha-plugin")
|
||||
neutral = FakeDistribution("neutral-plugin")
|
||||
|
||||
def entry_points(*, group):
|
||||
if group == MODULES_ENTRY_POINT_GROUP:
|
||||
return [_entry_point("zeta", zeta), _entry_point("alpha", alpha)]
|
||||
if group == IOS_CLI_PLUGIN_GROUP:
|
||||
return [_entry_point("zeta-ios", zeta)]
|
||||
if group == ANDROID_CLI_PLUGIN_GROUP:
|
||||
return [_entry_point("alpha-android", alpha)]
|
||||
if group == NEUTRAL_CLI_PLUGIN_GROUP:
|
||||
return [_entry_point("shared", neutral)]
|
||||
return []
|
||||
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.updates.importlib.metadata.entry_points", entry_points
|
||||
)
|
||||
|
||||
distributions = installed_plugin_distributions()
|
||||
|
||||
assert [distribution.name for distribution in distributions] == [
|
||||
"alpha-plugin",
|
||||
"neutral-plugin",
|
||||
"zeta-plugin",
|
||||
]
|
||||
|
||||
|
||||
def test_installed_plugin_distributions_skips_mvt_and_orphan_entry_points(monkeypatch):
|
||||
entry_points = [
|
||||
_entry_point("builtin", FakeDistribution("mvt")),
|
||||
SimpleNamespace(name="orphan", value="plugin:modules", dist=None),
|
||||
_entry_point("plugin", FakeDistribution("example-plugin")),
|
||||
]
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.updates.importlib.metadata.entry_points",
|
||||
lambda **kwargs: entry_points,
|
||||
)
|
||||
|
||||
distributions = installed_plugin_distributions()
|
||||
|
||||
assert [distribution.name for distribution in distributions] == ["example-plugin"]
|
||||
|
||||
|
||||
def test_installed_plugin_distributions_survives_broken_metadata(monkeypatch, caplog):
|
||||
def entry_points(*, group):
|
||||
raise RuntimeError("invalid package metadata")
|
||||
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.updates.importlib.metadata.entry_points", entry_points
|
||||
)
|
||||
|
||||
assert installed_plugin_distributions() == []
|
||||
assert "Unable to discover installed plugin packages" in caplog.text
|
||||
|
||||
|
||||
def test_index_plugin_update_is_reported(monkeypatch, data_folder, one_plugin):
|
||||
one_plugin(FakeDistribution("example-plugin", version="1.0.0"))
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.updates.requests.get",
|
||||
lambda url, **kwargs: FakeResponse(payload={"info": {"version": "1.2.0"}}),
|
||||
)
|
||||
|
||||
findings = PluginUpdates().check()
|
||||
|
||||
assert findings == [
|
||||
{
|
||||
"name": "example-plugin",
|
||||
"installed": "1.0.0",
|
||||
"latest": "1.2.0",
|
||||
"origin": "pypi",
|
||||
"upgrade_command": "pip install -U example-plugin",
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
def test_index_plugin_queries_the_package_index_with_the_configured_timeout(
|
||||
monkeypatch, data_folder, one_plugin
|
||||
):
|
||||
one_plugin(FakeDistribution("example-plugin"))
|
||||
requests_made = []
|
||||
|
||||
def get(url, **kwargs):
|
||||
requests_made.append((url, kwargs))
|
||||
return FakeResponse(payload={"info": {"version": "1.0.0"}})
|
||||
|
||||
monkeypatch.setattr("mvt.common.updates.requests.get", get)
|
||||
monkeypatch.setattr("mvt.common.updates.settings.NETWORK_TIMEOUT", 3)
|
||||
|
||||
PluginUpdates().check()
|
||||
|
||||
assert requests_made == [
|
||||
("https://pypi.org/pypi/example-plugin/json", {"timeout": 3})
|
||||
]
|
||||
|
||||
|
||||
def test_up_to_date_index_plugin_is_not_reported(monkeypatch, data_folder, one_plugin):
|
||||
one_plugin(FakeDistribution("example-plugin", version="1.2.0"))
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.updates.requests.get",
|
||||
lambda url, **kwargs: FakeResponse(payload={"info": {"version": "1.2.0"}}),
|
||||
)
|
||||
|
||||
assert PluginUpdates().check() == []
|
||||
|
||||
|
||||
def test_unpublished_plugin_is_skipped_silently(monkeypatch, data_folder, one_plugin):
|
||||
one_plugin(FakeDistribution("private-plugin"))
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.updates.requests.get",
|
||||
lambda url, **kwargs: FakeResponse(status_code=404),
|
||||
)
|
||||
|
||||
assert PluginUpdates().check() == []
|
||||
|
||||
|
||||
def test_repository_plugin_following_a_branch_is_reported(
|
||||
monkeypatch, data_folder, one_plugin
|
||||
):
|
||||
one_plugin(_git_distribution(requested_revision="main"))
|
||||
calls = []
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.updates.subprocess.run",
|
||||
_fake_git(stdout=f"{REMOTE_COMMIT}\trefs/heads/main\n", calls=calls),
|
||||
)
|
||||
monkeypatch.delenv("GIT_SSH_COMMAND", raising=False)
|
||||
|
||||
findings = PluginUpdates().check()
|
||||
|
||||
assert findings == [
|
||||
{
|
||||
"name": "example-plugin",
|
||||
"installed": "aaaaaaaa",
|
||||
"latest": "bbbbbbbb",
|
||||
"origin": "git",
|
||||
"upgrade_command": (
|
||||
f"pip install -U 'example-plugin @ git+{REPOSITORY_URL}@main'"
|
||||
),
|
||||
}
|
||||
]
|
||||
command, options = calls[0]
|
||||
assert command == ["git", "ls-remote", REPOSITORY_URL, "main"]
|
||||
assert options["env"]["GIT_TERMINAL_PROMPT"] == "0"
|
||||
# ssh asks the terminal for a passphrase or a host key unless it is told
|
||||
# not to, which git itself cannot prevent.
|
||||
assert options["env"]["GIT_SSH_COMMAND"] == (
|
||||
"ssh -o BatchMode=yes -o ConnectTimeout=10"
|
||||
)
|
||||
|
||||
|
||||
def test_batch_mode_options_come_before_the_configured_ssh_options(
|
||||
monkeypatch, data_folder, one_plugin
|
||||
):
|
||||
one_plugin(_git_distribution(requested_revision="main"))
|
||||
calls = []
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.updates.subprocess.run",
|
||||
_fake_git(stdout=f"{REMOTE_COMMIT}\trefs/heads/main\n", calls=calls),
|
||||
)
|
||||
monkeypatch.setenv("GIT_SSH_COMMAND", "ssh -o BatchMode=no -i /home/analyst/key")
|
||||
|
||||
PluginUpdates().check()
|
||||
|
||||
# ssh uses the first value it is given for a keyword, so an analyst asking
|
||||
# for prompts cannot bring them back, while their other options still
|
||||
# apply.
|
||||
assert calls[0][1]["env"]["GIT_SSH_COMMAND"] == (
|
||||
"ssh -o BatchMode=yes -o ConnectTimeout=10 -o BatchMode=no -i /home/analyst/key"
|
||||
)
|
||||
|
||||
|
||||
def test_repository_plugin_without_a_revision_follows_the_default_branch(
|
||||
monkeypatch, data_folder, one_plugin
|
||||
):
|
||||
one_plugin(_git_distribution())
|
||||
calls = []
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.updates.subprocess.run",
|
||||
_fake_git(stdout=f"{REMOTE_COMMIT}\tHEAD\n", calls=calls),
|
||||
)
|
||||
|
||||
findings = PluginUpdates().check()
|
||||
|
||||
assert calls[0][0] == ["git", "ls-remote", REPOSITORY_URL, "HEAD"]
|
||||
assert findings[0]["upgrade_command"] == (
|
||||
f"pip install -U 'example-plugin @ git+{REPOSITORY_URL}'"
|
||||
)
|
||||
|
||||
|
||||
def test_hostile_revision_cannot_inject_into_the_upgrade_command(
|
||||
monkeypatch, data_folder, one_plugin
|
||||
):
|
||||
revision = "main$(id)`id`;id"
|
||||
one_plugin(_git_distribution(requested_revision=revision))
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.updates.subprocess.run",
|
||||
_fake_git(stdout=f"{REMOTE_COMMIT}\trefs/heads/{revision}\n"),
|
||||
)
|
||||
|
||||
upgrade_command = PluginUpdates().check()[0]["upgrade_command"]
|
||||
|
||||
# Single quotes are the only quoting a shell does not expand anything in.
|
||||
assert upgrade_command == (
|
||||
f"pip install -U 'example-plugin @ git+{REPOSITORY_URL}@{revision}'"
|
||||
)
|
||||
assert shlex.split(upgrade_command) == [
|
||||
"pip",
|
||||
"install",
|
||||
"-U",
|
||||
f"example-plugin @ git+{REPOSITORY_URL}@{revision}",
|
||||
]
|
||||
|
||||
|
||||
def test_repository_plugin_with_an_option_like_url_is_skipped(
|
||||
monkeypatch, data_folder, one_plugin
|
||||
):
|
||||
one_plugin(
|
||||
FakeDistribution(
|
||||
"example-plugin",
|
||||
direct_url={
|
||||
"url": "--upload-pack=touch /tmp/mvt",
|
||||
"vcs_info": {"vcs": "git", "commit_id": INSTALLED_COMMIT},
|
||||
},
|
||||
)
|
||||
)
|
||||
calls = []
|
||||
monkeypatch.setattr("mvt.common.updates.subprocess.run", _fake_git(calls=calls))
|
||||
|
||||
assert PluginUpdates().check() == []
|
||||
assert calls == []
|
||||
|
||||
|
||||
def test_repository_plugin_with_an_option_like_revision_is_skipped(
|
||||
monkeypatch, data_folder, one_plugin
|
||||
):
|
||||
one_plugin(_git_distribution(requested_revision="--upload-pack=touch /tmp/mvt"))
|
||||
calls = []
|
||||
monkeypatch.setattr("mvt.common.updates.subprocess.run", _fake_git(calls=calls))
|
||||
|
||||
assert PluginUpdates().check() == []
|
||||
assert calls == []
|
||||
|
||||
|
||||
def test_repository_plugin_at_the_latest_commit_is_not_reported(
|
||||
monkeypatch, data_folder, one_plugin
|
||||
):
|
||||
one_plugin(_git_distribution(requested_revision="main"))
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.updates.subprocess.run",
|
||||
_fake_git(stdout=f"{INSTALLED_COMMIT}\trefs/heads/main\n"),
|
||||
)
|
||||
|
||||
assert PluginUpdates().check() == []
|
||||
|
||||
|
||||
def test_commit_pinned_repository_plugin_is_never_outdated(
|
||||
monkeypatch, data_folder, one_plugin
|
||||
):
|
||||
one_plugin(_git_distribution(requested_revision=INSTALLED_COMMIT))
|
||||
calls = []
|
||||
monkeypatch.setattr("mvt.common.updates.subprocess.run", _fake_git(calls=calls))
|
||||
|
||||
assert PluginUpdates().check() == []
|
||||
assert calls == []
|
||||
|
||||
|
||||
def test_short_commit_pinned_repository_plugin_is_never_outdated(
|
||||
monkeypatch, data_folder, one_plugin
|
||||
):
|
||||
one_plugin(_git_distribution(requested_revision=INSTALLED_COMMIT[:10]))
|
||||
calls = []
|
||||
monkeypatch.setattr("mvt.common.updates.subprocess.run", _fake_git(calls=calls))
|
||||
|
||||
assert PluginUpdates().check() == []
|
||||
assert calls == []
|
||||
|
||||
|
||||
def test_tag_pinned_repository_plugin_is_never_outdated(
|
||||
monkeypatch, data_folder, one_plugin
|
||||
):
|
||||
one_plugin(_git_distribution(requested_revision="v1.0.0"))
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.updates.subprocess.run",
|
||||
_fake_git(stdout=f"{REMOTE_COMMIT}\trefs/tags/v1.0.0\n"),
|
||||
)
|
||||
|
||||
assert PluginUpdates().check() == []
|
||||
|
||||
|
||||
def test_repository_plugin_is_skipped_without_git(monkeypatch, data_folder, one_plugin):
|
||||
one_plugin(_git_distribution(requested_revision="main"))
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.updates.subprocess.run",
|
||||
_fake_git(exception=FileNotFoundError("git")),
|
||||
)
|
||||
|
||||
assert PluginUpdates().check() == []
|
||||
|
||||
|
||||
def test_repository_plugin_is_skipped_when_git_fails(
|
||||
monkeypatch, data_folder, one_plugin
|
||||
):
|
||||
one_plugin(_git_distribution(requested_revision="main"))
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.updates.subprocess.run",
|
||||
_fake_git(stdout="", returncode=128),
|
||||
)
|
||||
|
||||
assert PluginUpdates().check() == []
|
||||
|
||||
|
||||
def test_local_plugin_install_is_skipped(monkeypatch, data_folder, one_plugin):
|
||||
one_plugin(
|
||||
FakeDistribution(
|
||||
"example-plugin",
|
||||
direct_url={
|
||||
"url": "file:///home/analyst/example-plugin",
|
||||
"dir_info": {"editable": True},
|
||||
},
|
||||
)
|
||||
)
|
||||
|
||||
def fail(*args, **kwargs):
|
||||
raise AssertionError("a local plugin install must not be checked")
|
||||
|
||||
monkeypatch.setattr("mvt.common.updates.requests.get", fail)
|
||||
monkeypatch.setattr("mvt.common.updates.subprocess.run", fail)
|
||||
|
||||
assert PluginUpdates().check() == []
|
||||
|
||||
|
||||
def test_check_stores_the_findings_and_the_check_timestamp(
|
||||
monkeypatch, data_folder, one_plugin
|
||||
):
|
||||
one_plugin(FakeDistribution("example-plugin", version="1.0.0"))
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.updates.requests.get",
|
||||
lambda url, **kwargs: FakeResponse(payload={"info": {"version": "1.2.0"}}),
|
||||
)
|
||||
plugin_updates = PluginUpdates()
|
||||
|
||||
findings = plugin_updates.check()
|
||||
|
||||
assert json.loads((data_folder / "plugin_updates.json").read_text()) == findings
|
||||
assert (data_folder / "latest_plugins_check").read_text().isdigit()
|
||||
assert PluginUpdates().get_findings() == findings
|
||||
|
||||
|
||||
def test_findings_are_empty_before_the_first_check(data_folder):
|
||||
assert PluginUpdates().get_findings() == []
|
||||
|
||||
|
||||
def test_malformed_cached_findings_are_dropped(data_folder):
|
||||
plugin_updates = PluginUpdates()
|
||||
usable = {
|
||||
"name": "example-plugin",
|
||||
"installed": "1.0.0",
|
||||
"latest": "1.2.0",
|
||||
"origin": "pypi",
|
||||
"upgrade_command": "pip install -U example-plugin",
|
||||
}
|
||||
data_folder.mkdir(parents=True, exist_ok=True)
|
||||
(data_folder / "plugin_updates.json").write_text(
|
||||
json.dumps(
|
||||
[
|
||||
{"oops": 1},
|
||||
"not a finding",
|
||||
{"name": "half-plugin", "installed": "1.0.0"},
|
||||
{**usable, "latest": None},
|
||||
usable,
|
||||
]
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
assert plugin_updates.get_findings() == [usable]
|
||||
|
||||
|
||||
def test_corrupt_cached_findings_are_ignored(data_folder):
|
||||
data_folder.mkdir(parents=True, exist_ok=True)
|
||||
(data_folder / "plugin_updates.json").write_text("{ not json", encoding="utf-8")
|
||||
|
||||
assert PluginUpdates().get_findings() == []
|
||||
|
||||
|
||||
def test_cached_findings_of_upgraded_and_removed_plugins_are_dropped(
|
||||
monkeypatch, data_folder
|
||||
):
|
||||
findings = [
|
||||
{
|
||||
"name": "upgraded-plugin",
|
||||
"installed": "1.0.0",
|
||||
"latest": "1.2.0",
|
||||
"origin": "pypi",
|
||||
"upgrade_command": "pip install -U upgraded-plugin",
|
||||
},
|
||||
{
|
||||
"name": "removed-plugin",
|
||||
"installed": "1.0.0",
|
||||
"latest": "1.2.0",
|
||||
"origin": "pypi",
|
||||
"upgrade_command": "pip install -U removed-plugin",
|
||||
},
|
||||
{
|
||||
"name": "example-plugin",
|
||||
"installed": "1.0.0",
|
||||
"latest": "1.2.0",
|
||||
"origin": "pypi",
|
||||
"upgrade_command": "pip install -U example-plugin",
|
||||
},
|
||||
]
|
||||
plugin_updates = PluginUpdates()
|
||||
plugin_updates.set_findings(findings)
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.updates.installed_plugin_distributions",
|
||||
lambda: [
|
||||
# The analyst upgraded this plugin since the latest check.
|
||||
FakeDistribution("upgraded-plugin", version="1.2.0"),
|
||||
FakeDistribution("example-plugin", version="1.0.0"),
|
||||
],
|
||||
)
|
||||
|
||||
assert plugin_updates.current_findings() == [findings[2]]
|
||||
|
||||
|
||||
def test_cached_findings_of_updated_repository_plugins_are_dropped(data_folder):
|
||||
findings = [
|
||||
{
|
||||
"name": "example-plugin",
|
||||
"installed": "aaaaaaaa",
|
||||
"latest": "bbbbbbbb",
|
||||
"origin": "git",
|
||||
"upgrade_command": "pip install -U example-plugin",
|
||||
}
|
||||
]
|
||||
plugin_updates = PluginUpdates()
|
||||
plugin_updates.set_findings(findings)
|
||||
|
||||
assert plugin_updates.current_findings([_git_distribution()]) == findings
|
||||
assert (
|
||||
plugin_updates.current_findings([_git_distribution(commit=REMOTE_COMMIT)]) == []
|
||||
)
|
||||
|
||||
|
||||
def test_corrupt_check_timestamp_does_not_raise(data_folder):
|
||||
plugin_updates = PluginUpdates()
|
||||
data_folder.mkdir(parents=True, exist_ok=True)
|
||||
(data_folder / "latest_plugins_check").write_text("truncated", encoding="utf-8")
|
||||
|
||||
assert plugin_updates.get_latest_check() == 0
|
||||
assert plugin_updates.should_check() == (True, 0)
|
||||
|
||||
|
||||
def test_should_check_is_throttled_for_twelve_hours(data_folder):
|
||||
plugin_updates = PluginUpdates()
|
||||
plugin_updates.set_findings([])
|
||||
|
||||
recent = datetime.now() - timedelta(hours=4)
|
||||
with open(plugin_updates.latest_check_path, "w", encoding="utf-8") as handle:
|
||||
handle.write(str(int(recent.timestamp())))
|
||||
|
||||
should_check, hours = plugin_updates.should_check()
|
||||
assert not should_check
|
||||
assert hours == 8
|
||||
|
||||
old = datetime.now() - timedelta(hours=13)
|
||||
with open(plugin_updates.latest_check_path, "w", encoding="utf-8") as handle:
|
||||
handle.write(str(int(old.timestamp())))
|
||||
|
||||
assert plugin_updates.should_check() == (True, 0)
|
||||
|
||||
|
||||
def test_should_check_without_a_previous_check(data_folder):
|
||||
assert PluginUpdates().should_check() == (True, 0)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def no_version_check(monkeypatch):
|
||||
monkeypatch.setattr(MVTUpdates, "check", lambda self: "")
|
||||
# Keep rich from wrapping the plugin lines while they are being asserted.
|
||||
monkeypatch.setenv("COLUMNS", "200")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def throttled_cache(monkeypatch, data_folder):
|
||||
"""Fill the findings cache and put the check inside its throttle window."""
|
||||
|
||||
def fill(findings, distributions):
|
||||
PluginUpdates().set_findings(findings)
|
||||
monkeypatch.setattr(
|
||||
logo, "installed_plugin_distributions", lambda: distributions
|
||||
)
|
||||
monkeypatch.setattr(PluginUpdates, "should_check", lambda self: (False, 8))
|
||||
monkeypatch.setattr(
|
||||
PluginUpdates,
|
||||
"check",
|
||||
lambda self: pytest.fail("the check must be throttled"),
|
||||
)
|
||||
|
||||
return fill
|
||||
|
||||
|
||||
def test_logo_prints_the_cached_plugin_updates(
|
||||
capsys, no_version_check, throttled_cache
|
||||
):
|
||||
throttled_cache(
|
||||
[
|
||||
{
|
||||
"name": "example-plugin",
|
||||
"installed": "1.0.0",
|
||||
"latest": "1.2.0",
|
||||
"origin": "pypi",
|
||||
"upgrade_command": "pip install -U example-plugin",
|
||||
}
|
||||
],
|
||||
[FakeDistribution("example-plugin", version="1.0.0")],
|
||||
)
|
||||
|
||||
logo.check_updates(disable_indicator_check=True)
|
||||
|
||||
output = capsys.readouterr().out
|
||||
assert "Plugin updates available:" in output
|
||||
assert "example-plugin 1.0.0 → 1.2.0 (pip install -U example-plugin)" in output
|
||||
|
||||
|
||||
def test_logo_does_not_print_a_cached_update_of_an_upgraded_plugin(
|
||||
capsys, no_version_check, throttled_cache
|
||||
):
|
||||
throttled_cache(
|
||||
[
|
||||
{
|
||||
"name": "example-plugin",
|
||||
"installed": "1.0.0",
|
||||
"latest": "1.2.0",
|
||||
"origin": "pypi",
|
||||
"upgrade_command": "pip install -U example-plugin",
|
||||
}
|
||||
],
|
||||
# The analyst already upgraded the plugin the cached finding is about.
|
||||
[FakeDistribution("example-plugin", version="1.2.0")],
|
||||
)
|
||||
|
||||
logo.check_updates(disable_indicator_check=True)
|
||||
|
||||
assert "Plugin updates" not in capsys.readouterr().out
|
||||
|
||||
|
||||
def test_logo_prints_nothing_when_throttled_without_findings(
|
||||
capsys, no_version_check, throttled_cache
|
||||
):
|
||||
throttled_cache([], [FakeDistribution("example-plugin")])
|
||||
|
||||
logo.check_updates(disable_indicator_check=True)
|
||||
|
||||
assert "Plugin updates" not in capsys.readouterr().out
|
||||
|
||||
|
||||
def test_logo_survives_a_corrupt_plugin_cache(
|
||||
monkeypatch, capsys, data_folder, no_version_check
|
||||
):
|
||||
data_folder.mkdir(parents=True, exist_ok=True)
|
||||
(data_folder / "plugin_updates.json").write_text(
|
||||
json.dumps([{"oops": 1}]), encoding="utf-8"
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
logo,
|
||||
"installed_plugin_distributions",
|
||||
lambda: [FakeDistribution("example-plugin")],
|
||||
)
|
||||
monkeypatch.setattr(PluginUpdates, "should_check", lambda self: (False, 8))
|
||||
|
||||
logo.check_updates(disable_indicator_check=True)
|
||||
|
||||
assert "Plugin updates" not in capsys.readouterr().out
|
||||
|
||||
|
||||
def test_logo_skips_the_plugin_check_without_plugins(
|
||||
monkeypatch, capsys, no_version_check
|
||||
):
|
||||
monkeypatch.setattr(logo, "installed_plugin_distributions", list)
|
||||
monkeypatch.setattr(
|
||||
PluginUpdates,
|
||||
"should_check",
|
||||
lambda self: pytest.fail("plugins must not be checked without plugins"),
|
||||
)
|
||||
|
||||
logo.check_updates(disable_indicator_check=True)
|
||||
|
||||
assert "Plugin updates" not in capsys.readouterr().out
|
||||
|
||||
|
||||
def test_logo_skips_the_plugin_check_without_network_access(
|
||||
monkeypatch, capsys, no_version_check
|
||||
):
|
||||
monkeypatch.setattr("mvt.common.logo.settings.NETWORK_ACCESS_ALLOWED", False)
|
||||
monkeypatch.setattr(
|
||||
logo,
|
||||
"installed_plugin_distributions",
|
||||
lambda: pytest.fail("plugins must not be listed without network access"),
|
||||
)
|
||||
|
||||
logo.check_updates(disable_indicator_check=True)
|
||||
|
||||
assert "Plugin updates" not in capsys.readouterr().out
|
||||
|
||||
|
||||
def test_logo_skips_the_plugin_check_when_update_checks_are_disabled(
|
||||
monkeypatch, capsys
|
||||
):
|
||||
monkeypatch.setattr(
|
||||
logo,
|
||||
"installed_plugin_distributions",
|
||||
lambda: pytest.fail("plugins must not be checked with --disable-update-check"),
|
||||
)
|
||||
|
||||
logo.check_updates(disable_version_check=True, disable_indicator_check=True)
|
||||
|
||||
assert capsys.readouterr().out == ""
|
||||
@@ -8,6 +8,11 @@ import os
|
||||
|
||||
import pytest
|
||||
|
||||
from mvt.common.cli_plugins import (
|
||||
MVT_ANDROID_CUSTOM_COMMANDS_ENV,
|
||||
MVT_CUSTOM_COMMANDS_ENV,
|
||||
MVT_IOS_CUSTOM_COMMANDS_ENV,
|
||||
)
|
||||
from mvt.common.indicators import Indicators
|
||||
|
||||
from .artifacts.generate_stix import generate_test_stix_file
|
||||
@@ -58,3 +63,33 @@ def indicators_factory(indicator_file):
|
||||
return ind
|
||||
|
||||
return f
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def restore_cli_commands(monkeypatch):
|
||||
"""Keep the external commands a test registers out of the next test.
|
||||
|
||||
Each CLI group is a module-level object shared by every test, so a test
|
||||
registering plugin or environment commands on one has to put it back. The
|
||||
groups are imported here rather than at the top of the file, so that
|
||||
collecting the tests does not import three CLIs for the sake of one
|
||||
fixture.
|
||||
"""
|
||||
from mvt.android.cli import cli as android_cli
|
||||
from mvt.cli import cli as neutral_cli
|
||||
from mvt.ios.cli import cli as ios_cli
|
||||
|
||||
groups = (neutral_cli, ios_cli, android_cli)
|
||||
for variable in (
|
||||
MVT_CUSTOM_COMMANDS_ENV,
|
||||
MVT_IOS_CUSTOM_COMMANDS_ENV,
|
||||
MVT_ANDROID_CUSTOM_COMMANDS_ENV,
|
||||
):
|
||||
monkeypatch.delenv(variable, raising=False)
|
||||
originals = [dict(group.commands) for group in groups]
|
||||
yield
|
||||
for group, commands in zip(groups, originals):
|
||||
group.commands.clear()
|
||||
group.commands.update(commands)
|
||||
if hasattr(group, "_mvt_external_command_sources"):
|
||||
delattr(group, "_mvt_external_command_sources")
|
||||
@@ -0,0 +1,49 @@
|
||||
# Mobile Verification Toolkit (MVT)
|
||||
# Copyright (c) 2021-2026 The MVT Authors.
|
||||
# Use of this software is governed by the MVT License 1.1 that can be found at
|
||||
# https://license.mvt.re/1.1/
|
||||
|
||||
from click.testing import CliRunner
|
||||
|
||||
from mvt.cli import cli
|
||||
from mvt.common.updates import IndicatorsUpdates
|
||||
from mvt.common.version import MVT_VERSION
|
||||
|
||||
# Keep the banner of the group callback from checking for updates online.
|
||||
OFFLINE = ["--disable-update-check", "--disable-indicator-update-check"]
|
||||
|
||||
|
||||
class TestMvtCommand:
|
||||
def test_running_mvt_alone_shows_the_logo_and_the_commands(self):
|
||||
result = CliRunner().invoke(cli, OFFLINE)
|
||||
|
||||
assert result.exit_code == 0
|
||||
logo_at = result.output.index("Mobile Verification Toolkit")
|
||||
usage_at = result.output.index("Usage:")
|
||||
assert logo_at < usage_at
|
||||
assert "mvt-ios" in result.output and "mvt-android" in result.output
|
||||
|
||||
def test_help_reminds_where_the_analysis_runs(self):
|
||||
result = CliRunner().invoke(cli, ["--help"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "mvt-ios" in result.output
|
||||
assert "mvt-android" in result.output
|
||||
assert "check-*" in result.output
|
||||
|
||||
def test_version_prints_the_installed_version(self):
|
||||
result = CliRunner().invoke(cli, [*OFFLINE, "version"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert f"Version: {MVT_VERSION}" in result.output
|
||||
|
||||
def test_download_iocs_updates_the_indicators(self, monkeypatch):
|
||||
updates = []
|
||||
monkeypatch.setattr(
|
||||
IndicatorsUpdates, "update", lambda self: updates.append(self)
|
||||
)
|
||||
|
||||
result = CliRunner().invoke(cli, [*OFFLINE, "download-iocs"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert len(updates) == 1
|
||||
@@ -10,10 +10,20 @@ import click
|
||||
import pytest
|
||||
|
||||
import mvt.android
|
||||
import mvt.cli
|
||||
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.cli import cli as mvt_cli
|
||||
from mvt.cli import main as mvt_main
|
||||
from mvt.common.cli_plugins import (
|
||||
ANDROID_CLI_PLUGIN_GROUP,
|
||||
IOS_CLI_PLUGIN_GROUP,
|
||||
MVT_ANDROID_CUSTOM_COMMANDS_ENV,
|
||||
MVT_CUSTOM_COMMANDS_ENV,
|
||||
MVT_IOS_CUSTOM_COMMANDS_ENV,
|
||||
NEUTRAL_CLI_PLUGIN_GROUP,
|
||||
)
|
||||
from mvt.ios.cli import cli as ios_cli
|
||||
from mvt.ios.cli import main as ios_main
|
||||
|
||||
@@ -39,23 +49,32 @@ def cli():
|
||||
"""
|
||||
|
||||
PROGRAMS = {
|
||||
"mvt-ios": (mvt.ios, ios_cli, IOS_CLI_PLUGIN_GROUP),
|
||||
"mvt-android": (mvt.android, android_cli, ANDROID_CLI_PLUGIN_GROUP),
|
||||
"mvt": (mvt.cli, mvt_cli, NEUTRAL_CLI_PLUGIN_GROUP, MVT_CUSTOM_COMMANDS_ENV),
|
||||
"mvt-ios": (mvt.ios, ios_cli, IOS_CLI_PLUGIN_GROUP, MVT_IOS_CUSTOM_COMMANDS_ENV),
|
||||
"mvt-android": (
|
||||
mvt.android,
|
||||
android_cli,
|
||||
ANDROID_CLI_PLUGIN_GROUP,
|
||||
MVT_ANDROID_CUSTOM_COMMANDS_ENV,
|
||||
),
|
||||
}
|
||||
|
||||
CASE_SUMMARY_COMMAND = """
|
||||
import click
|
||||
|
||||
@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")
|
||||
|
||||
@click.command("case-summary")
|
||||
def cli():
|
||||
click.echo("case summary ran")
|
||||
"""
|
||||
|
||||
# The entry-point group of another program, for each program: no group may add
|
||||
# its commands to a CLI other than its own.
|
||||
OTHER_PROGRAMS_GROUP = {
|
||||
"mvt": IOS_CLI_PLUGIN_GROUP,
|
||||
"mvt-ios": NEUTRAL_CLI_PLUGIN_GROUP,
|
||||
"mvt-android": NEUTRAL_CLI_PLUGIN_GROUP,
|
||||
}
|
||||
|
||||
|
||||
def _install_fixture_entry_point(monkeypatch, entry_point_group, command):
|
||||
@@ -92,7 +111,7 @@ def _offline_argv(program, *arguments):
|
||||
def test_main_registers_installed_plugins_before_running_the_cli(
|
||||
program, monkeypatch, capsys, restore_cli_commands
|
||||
):
|
||||
package, group, entry_point_group = PROGRAMS[program]
|
||||
package, group, entry_point_group, _ = PROGRAMS[program]
|
||||
|
||||
@click.command()
|
||||
def fixture_command():
|
||||
@@ -113,7 +132,7 @@ def test_main_registers_installed_plugins_before_running_the_cli(
|
||||
def test_main_completes_plugin_command_names(
|
||||
program, monkeypatch, capsys, restore_cli_commands
|
||||
):
|
||||
package, _, entry_point_group = PROGRAMS[program]
|
||||
package, _, entry_point_group, _ = PROGRAMS[program]
|
||||
|
||||
@click.command()
|
||||
def fixture_command():
|
||||
@@ -136,17 +155,9 @@ def test_main_completes_plugin_command_names(
|
||||
def test_main_still_loads_commands_from_a_file(
|
||||
program, monkeypatch, capsys, tmp_path, restore_cli_commands
|
||||
):
|
||||
package, _, entry_point_group = PROGRAMS[program]
|
||||
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",
|
||||
)
|
||||
command_path.write_text(CASE_SUMMARY_COMMAND, encoding="utf-8")
|
||||
_install_fixture_entry_point(
|
||||
monkeypatch, entry_point_group, click.Command("unused")
|
||||
)
|
||||
@@ -163,12 +174,66 @@ def test_main_still_loads_commands_from_a_file(
|
||||
assert "case summary ran" in capsys.readouterr().out
|
||||
|
||||
|
||||
@pytest.mark.parametrize("program", sorted(PROGRAMS))
|
||||
def test_main_loads_commands_from_the_environment_variable(
|
||||
program, monkeypatch, capsys, tmp_path, restore_cli_commands
|
||||
):
|
||||
# Each CLI reads its own variable, so a main() reading another CLI's would
|
||||
# go unnoticed without this.
|
||||
package, _, _, environment_variable = PROGRAMS[program]
|
||||
command_path = tmp_path / "case_summary.py"
|
||||
command_path.write_text(CASE_SUMMARY_COMMAND, encoding="utf-8")
|
||||
monkeypatch.setenv(environment_variable, str(command_path))
|
||||
monkeypatch.setattr(sys, "argv", _offline_argv(program, "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
|
||||
|
||||
|
||||
@pytest.mark.parametrize("program", sorted(PROGRAMS))
|
||||
def test_main_ignores_the_entry_point_groups_of_the_other_programs(
|
||||
program, monkeypatch, capsys, restore_cli_commands
|
||||
):
|
||||
package, group, _, _ = PROGRAMS[program]
|
||||
_install_fixture_entry_point(
|
||||
monkeypatch,
|
||||
OTHER_PROGRAMS_GROUP[program],
|
||||
click.Command(FIXTURE_COMMAND_NAME),
|
||||
)
|
||||
monkeypatch.setattr(sys, "argv", _offline_argv(program, "--help"))
|
||||
|
||||
with pytest.raises(SystemExit) as exit_info:
|
||||
package.main()
|
||||
|
||||
assert exit_info.value.code == 0
|
||||
assert FIXTURE_COMMAND_NAME not in group.commands
|
||||
assert FIXTURE_COMMAND_NAME not in capsys.readouterr().out
|
||||
|
||||
|
||||
def test_the_console_script_targets_are_importable():
|
||||
# [project.scripts] points at these, so they must stay on the packages.
|
||||
# [project.scripts] points at these, so they must stay where they are.
|
||||
assert mvt.cli.main is mvt_main
|
||||
assert mvt.ios.main is ios_main
|
||||
assert mvt.android.main is android_main
|
||||
|
||||
|
||||
def test_importing_mvt_does_not_import_a_cli(tmp_path):
|
||||
# The mvt package deliberately re-exports nothing of mvt.cli, so that
|
||||
# importing MVT stays cheap and free of side effects.
|
||||
result = run_isolated_python(
|
||||
"import sys\n"
|
||||
"import mvt\n"
|
||||
"print('imported a cli' if 'mvt.cli' in sys.modules else 'imported mvt')\n",
|
||||
home=tmp_path / "home",
|
||||
)
|
||||
|
||||
assert result.returncode == 0, result.stderr
|
||||
assert result.stdout.strip() == "imported mvt"
|
||||
|
||||
|
||||
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
|
||||
|
||||
+43
-23
@@ -6,56 +6,60 @@
|
||||
from click.testing import CliRunner
|
||||
|
||||
from mvt.android.cli import cli as android_cli
|
||||
from mvt.cli import cli as mvt_cli
|
||||
from mvt.ios.cli import cli as ios_cli
|
||||
|
||||
|
||||
class TestCompletionCommand:
|
||||
def test_completion_prints_instructions_by_default(self):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(ios_cli, ["completion"])
|
||||
result = runner.invoke(mvt_cli, ["completion"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "Shell completion for mvt-ios" in result.output
|
||||
assert "mvt-ios completion bash > ~/.mvt-ios-complete.bash" in result.output
|
||||
assert "Shell completion for mvt, mvt-ios and mvt-android" in result.output
|
||||
assert "mvt completion bash > ~/.mvt-complete.bash" in result.output
|
||||
assert "Mobile Verification Toolkit" not in result.output
|
||||
|
||||
def test_completion_prints_bash_script(self):
|
||||
def test_completion_bash_script_covers_every_cli(self):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(ios_cli, ["completion", "bash"])
|
||||
result = runner.invoke(mvt_cli, ["completion", "bash"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "_MVT_COMPLETE=bash_complete" in result.output
|
||||
assert "_MVT_IOS_COMPLETE=bash_complete" in result.output
|
||||
assert "_MVT_ANDROID_COMPLETE=bash_complete" in result.output
|
||||
assert "complete -o nosort" in result.output
|
||||
assert "mvt-ios" in result.output
|
||||
assert "Mobile Verification Toolkit" not in result.output
|
||||
|
||||
def test_completion_prints_fish_script(self):
|
||||
def test_completion_fish_script_covers_every_cli(self):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(android_cli, ["completion", "fish"])
|
||||
result = runner.invoke(mvt_cli, ["completion", "fish"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "_MVT_ANDROID_COMPLETE=fish_complete" in result.output
|
||||
assert "complete --no-files --command mvt-ios" in result.output
|
||||
assert "complete --no-files --command mvt-android" in result.output
|
||||
assert "complete --no-files --command mvt " in result.output
|
||||
assert "Mobile Verification Toolkit" not in result.output
|
||||
|
||||
def test_completion_install_updates_bashrc_once(self, tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HOME", str(tmp_path))
|
||||
runner = CliRunner()
|
||||
|
||||
result = runner.invoke(ios_cli, ["completion", "bash", "--install"])
|
||||
result = runner.invoke(mvt_cli, ["completion", "bash", "--install"])
|
||||
assert result.exit_code == 0
|
||||
|
||||
script_path = tmp_path / ".mvt-ios-complete.bash"
|
||||
script_path = tmp_path / ".mvt-complete.bash"
|
||||
bashrc_path = tmp_path / ".bashrc"
|
||||
assert script_path.exists()
|
||||
assert "_MVT_IOS_COMPLETE=bash_complete" in script_path.read_text(
|
||||
encoding="utf-8"
|
||||
)
|
||||
script = script_path.read_text(encoding="utf-8")
|
||||
assert "_MVT_COMPLETE=bash_complete" in script
|
||||
assert "_MVT_IOS_COMPLETE=bash_complete" in script
|
||||
assert "_MVT_ANDROID_COMPLETE=bash_complete" in script
|
||||
bashrc = bashrc_path.read_text(encoding="utf-8")
|
||||
assert "[ -f" in bashrc
|
||||
assert ".mvt-ios-complete.bash" in bashrc
|
||||
assert ".mvt-complete.bash" in bashrc
|
||||
|
||||
result = runner.invoke(ios_cli, ["completion", "bash", "--install"])
|
||||
result = runner.invoke(mvt_cli, ["completion", "bash", "--install"])
|
||||
assert result.exit_code == 0
|
||||
assert bashrc_path.read_text(encoding="utf-8") == bashrc
|
||||
|
||||
@@ -65,14 +69,30 @@ class TestCompletionCommand:
|
||||
monkeypatch.setenv("HOME", str(tmp_path))
|
||||
runner = CliRunner()
|
||||
|
||||
result = runner.invoke(android_cli, ["completion", "fish", "--install"])
|
||||
result = runner.invoke(mvt_cli, ["completion", "fish", "--install"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
script_path = (
|
||||
tmp_path / ".config" / "fish" / "completions" / "mvt-android.fish"
|
||||
)
|
||||
script_path = tmp_path / ".config" / "fish" / "conf.d" / "mvt-completion.fish"
|
||||
assert script_path.exists()
|
||||
assert "_MVT_ANDROID_COMPLETE=fish_complete" in script_path.read_text(
|
||||
encoding="utf-8"
|
||||
)
|
||||
script = script_path.read_text(encoding="utf-8")
|
||||
assert "_MVT_COMPLETE=fish_complete" in script
|
||||
assert "_MVT_IOS_COMPLETE=fish_complete" in script
|
||||
assert "_MVT_ANDROID_COMPLETE=fish_complete" in script
|
||||
assert not (tmp_path / ".fishrc").exists()
|
||||
assert not (tmp_path / ".bashrc").exists()
|
||||
assert not (tmp_path / ".zshrc").exists()
|
||||
|
||||
def test_completion_install_without_shell_is_a_usage_error(self):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(mvt_cli, ["completion", "--install"])
|
||||
|
||||
assert result.exit_code == 2
|
||||
assert "A shell is required when using --install." in result.output
|
||||
|
||||
def test_completion_is_not_a_command_of_the_platform_clis(self):
|
||||
runner = CliRunner()
|
||||
|
||||
assert "completion" not in ios_cli.commands
|
||||
assert "completion" not in android_cli.commands
|
||||
assert runner.invoke(ios_cli, ["completion"]).exit_code == 2
|
||||
assert runner.invoke(android_cli, ["completion"]).exit_code == 2
|
||||
Reference in new issue
Block a user