Compare commits

..
Author SHA1 Message Date
Janik Besendorf f5b7045b03 Merge main into feature/url-logging 2026-08-05 23:17:56 +02:00
Janik Besendorf c261316d7a Store message URLs in analysis output 2026-07-29 16:20:04 +02:00
21 changed files with 787 additions and 1895 deletions

No files matched your search

+1 -1
View File
@@ -11,7 +11,7 @@ jobs:
name: Add issue to project
runs-on: ubuntu-latest
steps:
- uses: actions/add-to-project@v2
- uses: actions/add-to-project@v0.5.0
with:
# You can target a project in a different organization
# to the issue
+3 -3
View File
@@ -11,13 +11,13 @@ jobs:
steps:
- name: Checkout
uses: actions/checkout@v7
uses: actions/checkout@v6
- name: Setup Python
uses: actions/setup-python@v7
uses: actions/setup-python@v6
with:
python-version: "3.10"
- name: Install uv
uses: astral-sh/setup-uv@v10.0.0
uses: astral-sh/setup-uv@v7
with:
enable-cache: true
- name: Install Dependencies
+6 -5
View File
@@ -36,10 +36,10 @@ jobs:
tag-suffix: "-android"
steps:
- name: Checkout repository
uses: actions/checkout@v7
uses: actions/checkout@v6
# Uses the `docker/login-action` action to log in to the Container registry registry using the account and password that will publish the packages. Once published, the packages are scoped to the account defined here.
- name: Log in to the Container registry
uses: docker/login-action@dbcb813823bdd20940b903addbd779551569679f # v4.6.0
uses: docker/login-action@65b78e6e13532edd9afa3aa52ac7964289d1a9c1
with:
registry: ${{ env.REGISTRY }}
username: ${{ github.actor }}
@@ -47,7 +47,7 @@ jobs:
# This step uses [docker/metadata-action](https://github.com/docker/metadata-action#about) to extract tags and labels that will be applied to the specified image. The `id` "meta" allows the output of this step to be referenced in a subsequent step. The `images` value provides the base name for the tags and labels.
- name: Extract metadata (tags, labels) for Docker
id: meta
uses: docker/metadata-action@v6
uses: docker/metadata-action@v5
with:
images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
flavor: |
@@ -61,13 +61,13 @@ jobs:
type=sha,format=long,suffix=${{ matrix.platform.tag-suffix }}
# This step sets up some additional capabilities to generate the provenance and sbom attestations
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v4
uses: docker/setup-buildx-action@v3
# This step uses the `docker/build-push-action` action to build the image, based on your repository's `Dockerfile`. If the build succeeds, it pushes the image to GitHub Packages.
# It uses the `context` parameter to define the build's context as the set of files located in the specified path. For more information, see "[Usage](https://github.com/docker/build-push-action#usage)" in the README of the `docker/build-push-action` repository.
# It uses the `tags` and `labels` parameters to tag and label the image with the output from the "meta" step.
- name: Build and push Docker image
id: push
uses: docker/build-push-action@v7
uses: docker/build-push-action@v6
with:
file: ${{ matrix.platform.dockerfile }}
context: .
@@ -76,3 +76,4 @@ jobs:
labels: ${{ steps.meta.outputs.labels }}
provenance: mode=max
sbom: true
+3 -3
View File
@@ -12,13 +12,13 @@ jobs:
steps:
- name: Checkout
uses: actions/checkout@v7
uses: actions/checkout@v6
- name: Setup Python
uses: actions/setup-python@v7
uses: actions/setup-python@v6
with:
python-version: "3.10"
- name: Install uv
uses: astral-sh/setup-uv@v10.0.0
uses: astral-sh/setup-uv@v7
with:
enable-cache: true
- name: Install Dependencies
+3 -3
View File
@@ -15,13 +15,13 @@ jobs:
python-version: ['3.10', '3.11', '3.12', '3.13', '3.14']
steps:
- uses: actions/checkout@v7
- uses: actions/checkout@v6
- name: Install uv
uses: astral-sh/setup-uv@v10.0.0
uses: astral-sh/setup-uv@v7
with:
enable-cache: true
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v7
uses: actions/setup-python@v6
with:
python-version: ${{ matrix.python-version }}
- name: Install Python dependencies
+3 -5
View File
@@ -11,12 +11,10 @@ jobs:
update-ios-version:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v7
- uses: actions/setup-python@v7
- name: Install script dependencies
run: python -m pip install "packaging==26.3"
- uses: actions/checkout@v3
- uses: actions/setup-python@v4
- name: Run script to fetch latest iOS releases from Apple RSS feed.
run: python .github/workflows/scripts/update-ios-releases.py
run: python3 .github/workflows/scripts/update-ios-releases.py
- name: Create Pull Request
uses: peter-evans/create-pull-request@v8
with:
-5
View File
@@ -82,11 +82,6 @@ Module-running `check-*` commands can load custom Python modules with
[development documentation](https://docs.mvt.re/en/latest/development/) for
details.
Users can also add top-level commands to `mvt-ios` and `mvt-android` from
installed Python packages or local files and folders. See the
[custom CLI command documentation](https://docs.mvt.re/en/latest/custom_commands/)
for the plugin entry points and `--load-command` interface.
## License
-107
View File
@@ -1,107 +0,0 @@
# Custom CLI Commands
MVT can load additional top-level commands into `mvt-ios` and `mvt-android`.
Custom commands are different from [custom forensic modules](development.md#custom-modules):
commands add new CLI operations, while modules add analysis steps to existing
`check-*` commands.
!!! warning
Custom commands run as trusted Python code inside the MVT process. Install
or load commands only from sources you trust. MVT does not sandbox
third-party commands, and the MVT maintainers do not maintain them.
## Install a Command Package
Python packages can register a Click command or group for either MVT CLI. A
minimal package can expose this command from `my_mvt_plugin.py`:
```python
import click
@click.command()
@click.argument("path", type=click.Path(exists=True))
def summarize(path):
"""Summarize an acquisition."""
click.echo(f"Summarizing {path}")
```
Register the object in the package's `pyproject.toml`. The entry-point name is
the command users invoke:
```toml
[project.entry-points."mvt.ios.cli_plugins"]
summarize = "my_mvt_plugin:summarize"
[project.entry-points."mvt.android.cli_plugins"]
summarize = "my_mvt_plugin:summarize"
```
Use only the iOS or Android group if the command is platform-specific. After
installing the package in the same environment as MVT, it appears directly in
the appropriate CLI:
```bash
mvt-ios summarize ./ios-backup
mvt-android summarize ./androidqf-output
```
For a `pipx` installation of MVT, inject the plugin into MVT's environment:
```bash
pipx inject mvt my-mvt-plugin
```
When MVT is installed in an active virtual environment, install the plugin with
`pip` in that environment.
## Load a Command File
For local commands that are not packaged, create a Python file that exports one
Click command or group named `cli`:
```python
import click
@click.command("case-summary")
@click.argument("path", type=click.Path(exists=True))
def cli(path):
"""Summarize a case directory."""
click.echo(f"Summarizing {path}")
```
Pass the file before the custom command name:
```bash
mvt-ios --load-command ./case_summary.py case-summary ./ios-backup
```
`--load-command` can be repeated and also accepts a folder. MVT loads
non-hidden top-level `*.py` files in sorted order and skips `__init__.py`.
Every loaded file must export one `cli` object.
To load a file or folder on every invocation, set the platform-specific
environment variable:
```bash
export MVT_IOS_CUSTOM_COMMANDS=./ios_commands
export MVT_ANDROID_CUSTOM_COMMANDS=./android_commands
```
## Naming and Errors
Built-in MVT commands cannot be replaced. External command names must also be
unique; when installed packages or environment paths collide, MVT keeps the
first command and logs a warning. A collision from an explicit
`--load-command` is a usage error.
A package entry point or environment command that cannot be imported appears
as a marked broken command without preventing other MVT commands from working.
Invoke that command to see its package or file source and the underlying error.
An invalid command supplied explicitly with `--load-command` fails immediately
with a usage error.
Installed command packages use the entry-point name as the CLI command name.
The entry point must resolve to a `click.Command` or `click.Group`.
-1
View File
@@ -31,7 +31,6 @@ nav:
- Introduction: "introduction.md"
- Installation: "install.md"
- Command Completion: "command_completion.md"
- Custom CLI Commands: "custom_commands.md"
- Using Docker: "docker.md"
- MVT for iOS:
- iOS Forensic Methodology: "ios/methodology.md"
+13 -13
View File
@@ -17,25 +17,25 @@ classifiers = [
"Programming Language :: Python",
]
dependencies = [
"click==8.4.2",
"rich==15.0.0",
"click==8.3.3",
"rich==14.3.3",
"tld==0.13.2",
"requests==2.34.2",
"simplejson==4.1.1",
"packaging==26.3",
"requests==2.33.1",
"simplejson==3.20.2",
"packaging==26.2",
"appdirs==1.4.4",
"iOSbackup==0.9.925",
"adb-shell[usb]==0.4.4",
"libusb1==3.4.0",
"cryptography==50.0.0",
"libusb1==3.3.1",
"cryptography==47.0.0",
"PyYAML>=6.0.2",
"pyahocorasick==2.3.1",
"betterproto2==0.10.0",
"pydantic==2.13.4",
"pydantic-settings==2.15.0",
"betterproto2==0.9.1",
"pydantic==2.13.3",
"pydantic-settings==2.14.0",
"NSKeyedUnArchiver==1.5.2",
"python-dateutil==2.9.0.post0",
"tzdata==2026.3",
"tzdata==2026.2",
]
requires-python = ">= 3.10"
@@ -65,9 +65,9 @@ dev = [
docs = [
"mkdocs==1.6.1",
"mkdocs-autorefs==1.4.4",
"mkdocs-material==9.7.7",
"mkdocs-material==9.7.6",
"mkdocs-material-extensions==1.3.1",
"mkdocstrings==1.0.6",
"mkdocstrings==1.0.4",
]
[build-system]
@@ -34,7 +34,6 @@ TOMBSTONE_TEXT_KEY_MAPPINGS = {
"signal": "signal_info",
"code": "code",
"Cause": "cause",
"Abort message": "abort_message",
}
@@ -68,8 +67,6 @@ class TombstoneCrashResult(pydantic.BaseModel):
uid: int
signal_info: SignalInfo
cause: Optional[str] = None
causes: Optional[List[dict]] = None
abort_message: Optional[str] = None
extra: Optional[str] = None
-14
View File
@@ -8,12 +8,6 @@ from zipfile import BadZipFile
import click
from mvt.common.cli_plugins import (
ANDROID_CLI_PLUGIN_GROUP,
MVT_ANDROID_CUSTOM_COMMANDS_ENV,
load_cli_commands_option,
register_cli_plugins,
)
from mvt.common.cmd_check_iocs import CmdCheckIOCS
from mvt.common.completion import (
SUPPORTED_SHELLS,
@@ -88,7 +82,6 @@ def _load_custom_modules(load_module):
# Main
# ==============================================================================
@click.group(invoke_without_command=False)
@load_cli_commands_option
@click.option(
"--disable-update-check", is_flag=True, help=HELP_MSG_DISABLE_UPDATE_CHECK
)
@@ -512,10 +505,3 @@ def check_iocs(ctx, iocs, list_modules, module, load_module, folder):
def download_indicators():
ioc_updates = IndicatorsUpdates()
ioc_updates.update()
register_cli_plugins(
cli,
entry_point_group=ANDROID_CLI_PLUGIN_GROUP,
environment_variable=MVT_ANDROID_CUSTOM_COMMANDS_ENV,
)
-297
View File
@@ -1,297 +0,0 @@
# Mobile Verification Toolkit (MVT)
# Copyright (c) 2021-2026 The MVT Authors.
# Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/
import hashlib
import importlib.metadata
import importlib.util
import logging
import os
import sys
from pathlib import Path
from types import ModuleType
from typing import Iterable
import click
IOS_CLI_PLUGIN_GROUP = "mvt.ios.cli_plugins"
ANDROID_CLI_PLUGIN_GROUP = "mvt.android.cli_plugins"
MVT_IOS_CUSTOM_COMMANDS_ENV = "MVT_IOS_CUSTOM_COMMANDS"
MVT_ANDROID_CUSTOM_COMMANDS_ENV = "MVT_ANDROID_CUSTOM_COMMANDS"
log = logging.getLogger(__name__)
class CustomCommandLoadError(Exception):
pass
class BrokenPluginCommand(click.Command):
"""A placeholder for an installed or configured command that failed to load."""
def __init__(self, name: str, source: str, exception: BaseException):
super().__init__(
name,
help=(
f"Unable to load external command from {source}.\n\n"
f"{type(exception).__name__}: {exception}"
),
short_help="Warning: external command could not be loaded.",
)
self.source = source
self.exception = exception
def parse_args(self, ctx: click.Context, args: list[str]) -> list[str]:
return args
def invoke(self, ctx: click.Context) -> None:
raise click.ClickException(
f"Unable to load external command '{self.name}' from {self.source}: "
f"{type(self.exception).__name__}: {self.exception}"
)
def _module_name_for_path(path: Path) -> str:
digest = hashlib.sha256(str(path).encode("utf-8")).hexdigest()[:16]
return f"_mvt_custom_command_{path.stem}_{digest}"
def _iter_command_files(path: Path) -> Iterable[Path]:
if path.is_file():
if path.suffix != ".py":
raise CustomCommandLoadError(
f"Custom command file is not a Python file: {path}"
)
yield path
return
if path.is_dir():
for child in sorted(path.iterdir()):
if child.name.startswith(".") or child.name == "__init__.py":
continue
if child.is_file() and child.suffix == ".py":
yield child
return
raise CustomCommandLoadError(f"Custom command path does not exist: {path}")
def _load_python_file(path: Path) -> ModuleType:
module_name = _module_name_for_path(path)
spec = importlib.util.spec_from_file_location(module_name, path)
if spec is None or spec.loader is None:
raise CustomCommandLoadError(f"Unable to load custom command file: {path}")
module = importlib.util.module_from_spec(spec)
sys.modules[module_name] = module
try:
spec.loader.exec_module(module)
except (Exception, SystemExit) as exc:
raise CustomCommandLoadError(
f"Unable to import custom command {path}: {exc}"
) from exc
return module
def load_cli_command_file(path: Path) -> click.Command:
module = _load_python_file(path)
command = getattr(module, "cli", None)
if not isinstance(command, click.Command):
raise CustomCommandLoadError(
f"Custom command {path} must export a Click command or group named 'cli'"
)
if command.name is None:
raise CustomCommandLoadError(f"Custom command {path} has no command name")
return command
def _register_command(
group: click.Group,
command: click.Command,
*,
name: str,
source: str,
collision_is_error: bool,
) -> bool:
if name in group.commands:
registered_sources = getattr(group, "_mvt_external_command_sources", {})
if registered_sources.get(name) == source:
return False
message = (
f"Unable to register external command '{name}' from {source}: "
"the command name is already registered"
)
if collision_is_error:
raise CustomCommandLoadError(message)
log.warning(message)
return False
group.add_command(command, name=name)
registered_sources = getattr(group, "_mvt_external_command_sources", {})
registered_sources[name] = source
setattr(group, "_mvt_external_command_sources", registered_sources)
return True
def register_cli_commands_from_path(
group: click.Group,
path: str | Path,
*,
collision_is_error: bool = False,
failures_are_errors: bool = False,
) -> list[str]:
resolved_path = Path(path).expanduser().resolve()
registered: list[str] = []
try:
command_files = list(_iter_command_files(resolved_path))
except CustomCommandLoadError as exc:
if failures_are_errors:
raise
name = resolved_path.stem.replace("_", "-") or "custom-command"
if _register_command(
group,
BrokenPluginCommand(name, str(resolved_path), exc),
name=name,
source=str(resolved_path),
collision_is_error=collision_is_error,
):
registered.append(name)
return registered
for command_file in command_files:
try:
command = load_cli_command_file(command_file)
except CustomCommandLoadError as exc:
if failures_are_errors:
raise
name = command_file.stem.replace("_", "-")
command = BrokenPluginCommand(name, str(command_file), exc)
command_name = command.name
if command_name is None:
raise CustomCommandLoadError(
f"Custom command {command_file} has no command name"
)
if _register_command(
group,
command,
name=command_name,
source=str(command_file),
collision_is_error=collision_is_error,
):
registered.append(command_name)
return registered
def _entry_point_source(entry_point: importlib.metadata.EntryPoint) -> str:
try:
distribution = getattr(entry_point, "dist", None)
if distribution is None:
return f"entry point {entry_point.value}"
name = distribution.metadata.get("Name", "unknown distribution")
version = getattr(distribution, "version", None)
if version:
return f"{name} {version} ({entry_point.value})"
return f"{name} ({entry_point.value})"
except Exception:
return f"entry point {entry_point.value}"
def register_installed_cli_commands(
group: click.Group,
entry_point_group: str,
) -> list[str]:
try:
entry_points = importlib.metadata.entry_points(group=entry_point_group)
except Exception as exc:
log.warning(
"Unable to discover external commands in entry-point group %s: %s",
entry_point_group,
exc,
)
return []
ordered_entry_points = sorted(
entry_points,
key=lambda entry_point: (
entry_point.name,
_entry_point_source(entry_point),
entry_point.value,
),
)
registered: list[str] = []
for entry_point in ordered_entry_points:
source = _entry_point_source(entry_point)
try:
command = entry_point.load()
if not isinstance(command, click.Command):
raise TypeError(
f"entry point must resolve to a Click command or group, "
f"not {type(command).__name__}"
)
except (Exception, SystemExit) as exc:
command = BrokenPluginCommand(entry_point.name, source, exc)
if _register_command(
group,
command,
name=entry_point.name,
source=source,
collision_is_error=False,
):
registered.append(entry_point.name)
return registered
def register_cli_plugins(
group: click.Group,
*,
entry_point_group: str,
environment_variable: str,
) -> None:
environment_path = os.environ.get(environment_variable)
if environment_path:
register_cli_commands_from_path(group, environment_path)
register_installed_cli_commands(group, entry_point_group)
def _load_command_option_callback(
ctx: click.Context,
param: click.Parameter,
paths: tuple[str, ...],
) -> tuple[str, ...]:
if not isinstance(ctx.command, click.Group):
raise click.ClickException("--load-command requires a Click command group")
for path in paths:
try:
register_cli_commands_from_path(
ctx.command,
path,
collision_is_error=True,
failures_are_errors=True,
)
except CustomCommandLoadError as exc:
raise click.BadParameter(str(exc), ctx=ctx, param=param) from exc
return paths
load_cli_commands_option = click.option(
"--load-command",
"load_commands",
type=click.Path(exists=True),
multiple=True,
expose_value=False,
is_eager=True,
callback=_load_command_option_callback,
help=(
"Load a custom CLI command from a Python file or folder "
"(can be invoked multiple times)"
),
)
-14
View File
@@ -8,12 +8,6 @@ import logging
import os
import click
from mvt.common.cli_plugins import (
IOS_CLI_PLUGIN_GROUP,
MVT_IOS_CUSTOM_COMMANDS_ENV,
load_cli_commands_option,
register_cli_plugins,
)
from mvt.common.cmd_check_iocs import CmdCheckIOCS
from mvt.common.completion import (
SUPPORTED_SHELLS,
@@ -92,7 +86,6 @@ def _load_custom_modules(load_module):
# Main
# ==============================================================================
@click.group(invoke_without_command=False)
@load_cli_commands_option
@click.option(
"--disable-update-check", is_flag=True, help=HELP_MSG_DISABLE_UPDATE_CHECK
)
@@ -530,10 +523,3 @@ def check_iocs(ctx, iocs, list_modules, module, load_module, folder):
def download_iocs():
ioc_updates = IndicatorsUpdates()
ioc_updates.update()
register_cli_plugins(
cli,
entry_point_group=IOS_CLI_PLUGIN_GROUP,
environment_variable=MVT_IOS_CUSTOM_COMMANDS_ENV,
)
+12 -66
View File
@@ -6,10 +6,8 @@
import json
import logging
import os
import shutil
import tarfile
from pathlib import Path, PurePosixPath
from tempfile import TemporaryDirectory
from pathlib import Path
from typing import Any, Optional
from mvt.common.command import Command
@@ -56,8 +54,6 @@ class CmdIOSCheckSysdiagnose(Command):
self.sysdiagnose_archive: Optional[tarfile.TarFile] = None
self.sysdiagnose_files: list[str] = []
self.ips_files: list[dict[str, Any]] = []
self.temp_sysdiagnose_dir: Optional[TemporaryDirectory[str]] = None
self.extracted_sysdiagnose_path: Optional[str] = None
@staticmethod
def _parse_bugtype_header(data: bytes) -> Optional[int]:
@@ -95,68 +91,21 @@ class CmdIOSCheckSysdiagnose(Command):
self.log.info("Parsing sysdiagnose archive. This might take a while...")
self.sysdiagnose_format = "tar"
self.sysdiagnose_archive = tarfile.open(self.target_path, "r:gz")
self._extract_sysdiagnose_archive()
def _extract_sysdiagnose_archive(self) -> None:
archive = self.sysdiagnose_archive
if archive is None:
raise RuntimeError("Sysdiagnose archive has not been initialized")
self.temp_sysdiagnose_dir = TemporaryDirectory()
extraction_root = Path(self.temp_sysdiagnose_dir.name).resolve()
archive_roots = set()
for member in archive:
member_path = PurePosixPath(member.name.replace("\\", "/"))
if member_path.is_absolute() or ".." in member_path.parts:
self.log.warning("Skipping unsafe sysdiagnose path %r", member.name)
continue
destination = extraction_root.joinpath(*member_path.parts).resolve()
if not destination.is_relative_to(extraction_root):
self.log.warning("Skipping unsafe sysdiagnose path %r", member.name)
continue
if not member_path.parts:
continue
archive_roots.add(member_path.parts[0])
if member.isdir():
destination.mkdir(parents=True, exist_ok=True)
continue
# Modules only need directories and regular files. Do not materialize
# links or device nodes from an untrusted sysdiagnose archive.
if not member.isfile():
self.log.warning("Skipping unsafe sysdiagnose member %r", member.name)
continue
normalized_name = member_path.as_posix()
self.sysdiagnose_files.append(normalized_name)
source = archive.extractfile(member)
if source is None:
continue
destination.parent.mkdir(parents=True, exist_ok=True)
with source, destination.open("wb") as output:
shutil.copyfileobj(source, output)
if normalized_name.endswith(".ips"):
self._add_ips_file(str(destination), destination.read_bytes())
if len(archive_roots) != 1:
raise ValueError("Sysdiagnose archive must contain one top-level directory")
self.extracted_sysdiagnose_path = str(extraction_root / archive_roots.pop())
for member in self.sysdiagnose_archive:
self.sysdiagnose_files.append(member.name)
if member.isfile() and member.name.endswith(".ips"):
archive_handle = self.sysdiagnose_archive.extractfile(member)
if archive_handle is not None:
with archive_handle:
self._add_ips_file(member.name, archive_handle.read())
def module_init(self, module) -> None:
module.ips_files = self.ips_files
if self.sysdiagnose_format == "tar":
if self.extracted_sysdiagnose_path is None:
raise RuntimeError("Sysdiagnose archive has not been extracted")
module.from_sysdiagnose_folder(
self.extracted_sysdiagnose_path, self.sysdiagnose_files
if self.sysdiagnose_archive is None:
raise RuntimeError("Sysdiagnose archive has not been initialized")
module.from_sysdiagnose_tar(
self.sysdiagnose_archive, self.sysdiagnose_files
)
return
if self.sysdiagnose_format == "dir" and self.target_path:
@@ -168,6 +117,3 @@ class CmdIOSCheckSysdiagnose(Command):
if self.sysdiagnose_archive is not None:
self.sysdiagnose_archive.close()
self.sysdiagnose_archive = None
if self.temp_sysdiagnose_dir is not None:
self.temp_sysdiagnose_dir.cleanup()
self.temp_sysdiagnose_dir = None
-8
View File
@@ -1216,10 +1216,6 @@
"version": "18.7.9",
"build": "22H355"
},
{
"version": "18.7.10",
"build": "22H373"
},
{
"version": "26",
"build": "23A341"
@@ -1259,9 +1255,5 @@
{
"version": "26.5",
"build": "23F77"
},
{
"version": "26.6.1",
"build": "23G82"
}
]
+1 -3
View File
@@ -73,9 +73,7 @@ class Calls(IOSExtraction):
"duration": row[1],
"location": row[2],
"number": (
row[3].decode("utf-8")
if isinstance(row[3], bytes)
else row[3]
row[3].decode("utf-8") if row[3] and row[3] is bytes else row[3]
),
"provider": row[4],
}
-42
View File
@@ -8,7 +8,6 @@ import datetime
import pytest
from mvt.android.artifacts.tombstone_crashes import TombstoneCrashArtifact
from mvt.android.parsers.proto.tombstone import Tombstone
from ..utils import get_artifact
@@ -43,47 +42,6 @@ class TestTombstoneCrashArtifact:
assert len(tombstone_artifact.results) == 1
self.validate_tombstone_result(tombstone_artifact.results[0])
def test_text_tombstone_preserves_abort_message(self):
tombstone_artifact = TombstoneCrashArtifact()
artifact_path = "android_data/bugreport/FS/data/tombstones/tombstone_00"
file = get_artifact(artifact_path)
with open(file, "rb") as f:
data = f.read()
tombstone_artifact.parse(
os.path.basename(artifact_path),
datetime.datetime(2021, 9, 29, 17, 43, 49),
data,
)
assert tombstone_artifact.results[0]["abort_message"] == (
"Check failed: payload.size() <= bytes_left "
"(payload.size()=99, bytes_left=51) "
)
def test_protobuf_tombstone_preserves_abort_message_and_causes(self):
tombstone_artifact = TombstoneCrashArtifact()
artifact_path = "android_data/tombstone_process.pb"
file = get_artifact(artifact_path)
with open(file, "rb") as f:
tombstone = Tombstone().parse(f.read())
tombstone.abort_message = "synthetic abort reason"
tombstone_artifact.parse_protobuf(
os.path.basename(artifact_path),
datetime.datetime(2023, 4, 12, 12, 32, 40, 518290),
bytes(tombstone),
)
result = tombstone_artifact.results[0]
assert result["abort_message"] == "synthetic abort reason"
assert result["causes"] == [
{
"human_readable": "null pointer dereference",
"memory_error": None,
}
]
def test_text_tombstone_keeps_crashing_thread(self):
tombstone_artifact = TombstoneCrashArtifact()
artifact_path = "android_data/tombstone_process.txt"
-371
View File
@@ -1,371 +0,0 @@
from types import SimpleNamespace
import click
from click.testing import CliRunner
from mvt.common.cli_plugins import (
ANDROID_CLI_PLUGIN_GROUP,
IOS_CLI_PLUGIN_GROUP,
BrokenPluginCommand,
load_cli_commands_option,
register_cli_commands_from_path,
register_cli_plugins,
register_installed_cli_commands,
)
COMMAND_TEMPLATE = """
import click
@click.command({name!r})
@click.pass_context
def cli(ctx):
click.echo({message!r})
if ctx.obj:
click.echo(ctx.obj.get("marker", ""))
"""
def _write_command(path, name, message="command ran"):
path.write_text(
COMMAND_TEMPLATE.format(name=name, message=message),
encoding="utf-8",
)
return path
def _make_group():
@click.group()
@load_cli_commands_option
@click.pass_context
def group(ctx):
ctx.ensure_object(dict)
ctx.obj["marker"] = "parent context"
return group
def _entry_point(name, value, command=None, exception=None, distribution="plugin"):
def load():
if exception is not None:
raise exception
return command
dist = SimpleNamespace(metadata={"Name": distribution}, version="1.0")
return SimpleNamespace(name=name, value=value, load=load, dist=dist)
def test_load_command_option_registers_command_before_resolution(tmp_path):
command_path = _write_command(tmp_path / "hello.py", "hello")
group = _make_group()
result = CliRunner().invoke(
group,
["--load-command", str(command_path), "hello"],
)
assert result.exit_code == 0
assert "command ran" in result.output
assert "parent context" in result.output
def test_load_command_option_supports_folders_and_repeated_paths(tmp_path):
folder = tmp_path / "commands"
folder.mkdir()
_write_command(folder / "b.py", "second")
_write_command(folder / "a.py", "first")
_write_command(folder / ".hidden.py", "hidden")
_write_command(folder / "__init__.py", "init")
other = _write_command(tmp_path / "third.py", "third")
group = _make_group()
result = CliRunner().invoke(
group,
[
"--load-command",
str(folder),
"--load-command",
str(other),
"--load-command",
str(other),
"--help",
],
)
assert result.exit_code == 0
assert "first" in result.output
assert "second" in result.output
assert "third" in result.output
assert "hidden" not in result.output
assert "init" not in result.output
def test_loaded_command_participates_in_shell_completion(tmp_path):
command_path = _write_command(tmp_path / "hello.py", "hello")
group = _make_group()
words = f"group --load-command {command_path} he"
result = CliRunner().invoke(
group,
[],
env={
"_GROUP_COMPLETE": "bash_complete",
"COMP_WORDS": words,
"COMP_CWORD": "3",
},
)
assert result.exit_code == 0
assert "plain,hello" in result.output
def test_explicit_command_import_and_contract_failures_are_usage_errors(tmp_path):
broken_path = tmp_path / "broken.py"
broken_path.write_text("raise RuntimeError('broken import')", encoding="utf-8")
missing_cli_path = tmp_path / "missing_cli.py"
missing_cli_path.write_text("value = 1", encoding="utf-8")
broken_result = CliRunner().invoke(
_make_group(),
["--load-command", str(broken_path), "broken"],
)
missing_cli_result = CliRunner().invoke(
_make_group(),
["--load-command", str(missing_cli_path), "missing-cli"],
)
assert broken_result.exit_code == 2
assert "broken import" in broken_result.output
assert missing_cli_result.exit_code == 2
assert "must export a Click command or group named 'cli'" in (
missing_cli_result.output
)
def test_environment_command_failure_gets_broken_placeholder(tmp_path):
command_path = tmp_path / "broken_command.py"
command_path.write_text("raise RuntimeError('broken import')", encoding="utf-8")
group = click.Group()
registered = register_cli_commands_from_path(group, command_path)
assert registered == ["broken-command"]
assert isinstance(group.commands["broken-command"], BrokenPluginCommand)
result = CliRunner().invoke(group, ["broken-command"])
assert result.exit_code == 1
assert "broken import" in result.output
assert str(command_path) in result.output
def test_environment_command_system_exit_gets_broken_placeholder(tmp_path):
command_path = tmp_path / "exiting_command.py"
command_path.write_text("raise SystemExit(7)", encoding="utf-8")
group = click.Group()
registered = register_cli_commands_from_path(group, command_path)
assert registered == ["exiting-command"]
assert isinstance(group.commands["exiting-command"], BrokenPluginCommand)
result = CliRunner().invoke(group, ["exiting-command"])
assert result.exit_code == 1
assert "Unable to import custom command" in result.output
assert result.output.rstrip().endswith(": 7")
def test_installed_entry_point_name_is_the_command_name(monkeypatch):
@click.command("internal-name")
def command():
click.echo("installed command ran")
entry_point = _entry_point(
"external-name",
"example_plugin:cli",
command=command,
)
monkeypatch.setattr(
"mvt.common.cli_plugins.importlib.metadata.entry_points",
lambda **kwargs: [entry_point],
)
group = click.Group()
registered = register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
assert registered == ["external-name"]
assert "internal-name" not in group.commands
result = CliRunner().invoke(group, ["external-name"])
assert result.exit_code == 0
assert result.output == "installed command ran\n"
def test_broken_installed_plugin_does_not_break_cli(monkeypatch):
broken = _entry_point(
"broken",
"broken_plugin:cli",
exception=RuntimeError("missing dependency"),
distribution="broken-plugin",
)
monkeypatch.setattr(
"mvt.common.cli_plugins.importlib.metadata.entry_points",
lambda **kwargs: [broken],
)
group = click.Group()
register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
help_result = CliRunner().invoke(group, ["--help"])
assert help_result.exit_code == 0
assert "Warning: external command could not be loaded." in help_result.output
result = CliRunner().invoke(group, ["broken"])
assert result.exit_code == 1
assert "broken-plugin 1.0 (broken_plugin:cli)" in result.output
assert "RuntimeError: missing dependency" in result.output
def test_installed_plugin_system_exit_does_not_break_cli(monkeypatch):
exiting = _entry_point(
"exiting",
"exiting_plugin:cli",
exception=SystemExit(7),
distribution="exiting-plugin",
)
monkeypatch.setattr(
"mvt.common.cli_plugins.importlib.metadata.entry_points",
lambda **kwargs: [exiting],
)
group = click.Group()
register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
help_result = CliRunner().invoke(group, ["--help"])
assert help_result.exit_code == 0
result = CliRunner().invoke(group, ["exiting"])
assert result.exit_code == 1
assert "SystemExit: 7" in result.output
def test_non_click_entry_point_gets_broken_placeholder(monkeypatch):
invalid = _entry_point("invalid", "plugin:value", command=object())
monkeypatch.setattr(
"mvt.common.cli_plugins.importlib.metadata.entry_points",
lambda **kwargs: [invalid],
)
group = click.Group()
register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
assert isinstance(group.commands["invalid"], BrokenPluginCommand)
result = CliRunner().invoke(group, ["invalid"])
assert result.exit_code == 1
assert "must resolve to a Click command or group" in result.output
def test_entry_point_discovery_failure_does_not_break_group(monkeypatch, caplog):
def fail_discovery(**kwargs):
raise RuntimeError("invalid package metadata")
monkeypatch.setattr(
"mvt.common.cli_plugins.importlib.metadata.entry_points",
fail_discovery,
)
group = click.Group()
registered = register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
assert registered == []
assert not group.commands
assert "Unable to discover external commands" in caplog.text
assert "invalid package metadata" in caplog.text
def test_builtin_and_first_external_command_win_collisions(monkeypatch, caplog):
@click.command("version")
def core_version():
pass
@click.command()
def first():
pass
@click.command()
def second():
pass
entry_points = [
_entry_point("duplicate", "z_plugin:cli", command=second, distribution="z"),
_entry_point("version", "plugin:version", command=first),
_entry_point("duplicate", "a_plugin:cli", command=first, distribution="a"),
]
monkeypatch.setattr(
"mvt.common.cli_plugins.importlib.metadata.entry_points",
lambda **kwargs: entry_points,
)
group = click.Group(commands={"version": core_version})
registered = register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
assert registered == ["duplicate"]
assert group.commands["version"] is core_version
assert group.commands["duplicate"] is first
assert "the command name is already registered" in caplog.text
def test_explicit_command_cannot_replace_existing_command(tmp_path):
command_path = _write_command(tmp_path / "version.py", "version")
group = _make_group()
@group.command("version")
def core_version():
pass
result = CliRunner().invoke(
group,
["--load-command", str(command_path), "version"],
)
assert result.exit_code == 2
assert "the command name is already registered" in result.output
def test_platform_entry_point_groups_and_environment_paths_are_separate(
tmp_path, monkeypatch
):
ios_path = _write_command(tmp_path / "ios.py", "ios-file")
android_path = _write_command(tmp_path / "android.py", "android-file")
@click.command()
def ios_package():
pass
@click.command()
def android_package():
pass
def entry_points(*, group):
if group == IOS_CLI_PLUGIN_GROUP:
return [_entry_point("ios-package", "ios_plugin:cli", ios_package)]
return [_entry_point("android-package", "android_plugin:cli", android_package)]
monkeypatch.setattr(
"mvt.common.cli_plugins.importlib.metadata.entry_points",
entry_points,
)
monkeypatch.setenv("TEST_IOS_COMMANDS", str(ios_path))
monkeypatch.setenv("TEST_ANDROID_COMMANDS", str(android_path))
ios_group = click.Group()
android_group = click.Group()
register_cli_plugins(
ios_group,
entry_point_group=IOS_CLI_PLUGIN_GROUP,
environment_variable="TEST_IOS_COMMANDS",
)
register_cli_plugins(
android_group,
entry_point_group=ANDROID_CLI_PLUGIN_GROUP,
environment_variable="TEST_ANDROID_COMMANDS",
)
assert set(ios_group.commands) == {"ios-file", "ios-package"}
assert set(android_group.commands) == {"android-file", "android-package"}
+1 -43
View File
@@ -1,7 +1,5 @@
import io
import tarfile
from datetime import timedelta
from pathlib import Path
from mvt.ios.cmd_check_sysdiagnose import CmdIOSCheckSysdiagnose
from mvt.ios.modules.sysdiagnose import SysdiagnoseExtraction
@@ -72,46 +70,6 @@ def test_check_sysdiagnose_from_archive_closes_archive(tmp_path):
{"content": "artifact", "timezone_offset": timedelta(hours=2).seconds}
]
assert command.executed[0].ips_files == [
{
"file_path": str(
Path(command.extracted_sysdiagnose_path) / "report.ips"
),
"bug_type": 210,
}
{"file_path": "sysdiagnose/report.ips", "bug_type": 210}
]
assert command.sysdiagnose_archive is None
def test_archive_is_extracted_once_and_unsafe_members_are_skipped(tmp_path):
archive_path = tmp_path / "sysdiagnose.tar.gz"
escaped_path = tmp_path / "escaped.txt"
content = b"test content"
member = tarfile.TarInfo("sysdiagnose/artifact.txt")
member.size = len(content)
with tarfile.open(archive_path, "w:gz") as archive:
archive.addfile(member, io.BytesIO(content))
escaped = tarfile.TarInfo(f"sysdiagnose/../../{escaped_path.name}")
escaped.size = len(content)
archive.addfile(escaped, io.BytesIO(content))
link = tarfile.TarInfo("sysdiagnose/link")
link.type = tarfile.SYMTYPE
link.linkname = "/etc/hostname"
archive.addfile(link)
command = CmdIOSCheckSysdiagnose(target_path=str(archive_path))
try:
command.init()
extracted_path = Path(command.extracted_sysdiagnose_path)
assert (extracted_path / "artifact.txt").read_bytes() == content
assert not escaped_path.exists()
assert not (extracted_path / "link").exists()
module = SysdiagnoseExtraction()
command.module_init(module)
assert module.tar is None
assert module.parent_path == str(extracted_path.parent)
finally:
command.finish()
assert not extracted_path.exists()
Generated
+741 -888
View File
File diff suppressed because it is too large. Load diff