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_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 == ""