diff --git a/README.md b/README.md index 5f00600..54cccca 100644 --- a/README.md +++ b/README.md @@ -58,7 +58,7 @@ For alternative installation options and known issues, please refer to the [docu ## Usage -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` 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/) +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 diff --git a/docs/development/custom_commands.md b/docs/development/custom_commands.md index 19cdb85..71ac623 100644 --- a/docs/development/custom_commands.md +++ b/docs/development/custom_commands.md @@ -60,7 +60,8 @@ 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) @@ -69,9 +70,9 @@ rather than in MVT's own `config.yaml`. ### Commands on `mvt` The `mvt` command hosts what belongs to neither platform: `version`, -`completion` 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 +`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 diff --git a/docs/development/index.md b/docs/development/index.md index c2b8464..692a2ab 100644 --- a/docs/development/index.md +++ b/docs/development/index.md @@ -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 diff --git a/docs/development/plugins.md b/docs/development/plugins.md new file mode 100644 index 0000000..0085975 --- /dev/null +++ b/docs/development/plugins.md @@ -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+` 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+"` 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. diff --git a/mkdocs.yml b/mkdocs.yml index 6250cdc..66def17 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -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" diff --git a/src/mvt/cli.py b/src/mvt/cli.py index a2e48dd..3ba6602 100644 --- a/src/mvt/cli.py +++ b/src/mvt/cli.py @@ -11,6 +11,7 @@ from mvt.common.cli_plugins import ( 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, @@ -84,6 +85,11 @@ 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 # ============================================================================== diff --git a/src/mvt/common/cmd_plugins.py b/src/mvt/common/cmd_plugins.py new file mode 100644 index 0000000..21619eb --- /dev/null +++ b/src/mvt/common/cmd_plugins.py @@ -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." + ) diff --git a/src/mvt/common/help.py b/src/mvt/common/help.py index 4eccef1..13575e5 100644 --- a/src/mvt/common/help.py +++ b/src/mvt/common/help.py @@ -24,6 +24,11 @@ HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK = "Disable indicators update check" 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" diff --git a/src/mvt/common/logo.py b/src/mvt/common/logo.py index 862479f..96836a9 100644 --- a/src/mvt/common/logo.py +++ b/src/mvt/common/logo.py @@ -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,68 +134,16 @@ 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[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]" - ) + _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[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.") + # 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( diff --git a/src/mvt/common/module_loader.py b/src/mvt/common/module_loader.py index 956da4c..37f447d 100644 --- a/src/mvt/common/module_loader.py +++ b/src/mvt/common/module_loader.py @@ -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 diff --git a/src/mvt/common/updates.py b/src/mvt/common/updates.py index 001a5c2..71e97cc 100644 --- a/src/mvt/common/updates.py +++ b/src/mvt/common/updates.py @@ -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: diff --git a/tests/common/test_cmd_plugins.py b/tests/common/test_cmd_plugins.py new file mode 100644 index 0000000..adf9322 --- /dev/null +++ b/tests/common/test_cmd_plugins.py @@ -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 diff --git a/tests/common/test_plugin_updates.py b/tests/common/test_plugin_updates.py new file mode 100644 index 0000000..9c45cea --- /dev/null +++ b/tests/common/test_plugin_updates.py @@ -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 == ""