mirror of
https://github.com/mvt-project/mvt.git
synced 2026-10-30 17:56:25 +01:00
Compare commits
7
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3f83d9ba6f | ||
|
|
47fd771b2a | ||
|
|
8123db26d9 | ||
|
|
d92a60c9be | ||
|
|
0b48d9fe1d | ||
|
|
067f053627 | ||
|
|
93b7fb5232 |
No files matched your search
@@ -11,7 +11,7 @@ jobs:
|
|||||||
name: Add issue to project
|
name: Add issue to project
|
||||||
runs-on: ubuntu-latest
|
runs-on: ubuntu-latest
|
||||||
steps:
|
steps:
|
||||||
- uses: actions/add-to-project@v0.5.0
|
- uses: actions/add-to-project@v2
|
||||||
with:
|
with:
|
||||||
# You can target a project in a different organization
|
# You can target a project in a different organization
|
||||||
# to the issue
|
# to the issue
|
||||||
|
|||||||
@@ -11,13 +11,13 @@ jobs:
|
|||||||
|
|
||||||
steps:
|
steps:
|
||||||
- name: Checkout
|
- name: Checkout
|
||||||
uses: actions/checkout@v6
|
uses: actions/checkout@v7
|
||||||
- name: Setup Python
|
- name: Setup Python
|
||||||
uses: actions/setup-python@v6
|
uses: actions/setup-python@v7
|
||||||
with:
|
with:
|
||||||
python-version: "3.10"
|
python-version: "3.10"
|
||||||
- name: Install uv
|
- name: Install uv
|
||||||
uses: astral-sh/setup-uv@v7
|
uses: astral-sh/setup-uv@v10.0.0
|
||||||
with:
|
with:
|
||||||
enable-cache: true
|
enable-cache: true
|
||||||
- name: Install Dependencies
|
- name: Install Dependencies
|
||||||
|
|||||||
@@ -36,10 +36,10 @@ jobs:
|
|||||||
tag-suffix: "-android"
|
tag-suffix: "-android"
|
||||||
steps:
|
steps:
|
||||||
- name: Checkout repository
|
- name: Checkout repository
|
||||||
uses: actions/checkout@v6
|
uses: actions/checkout@v7
|
||||||
# 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.
|
# 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
|
- name: Log in to the Container registry
|
||||||
uses: docker/login-action@65b78e6e13532edd9afa3aa52ac7964289d1a9c1
|
uses: docker/login-action@dbcb813823bdd20940b903addbd779551569679f # v4.6.0
|
||||||
with:
|
with:
|
||||||
registry: ${{ env.REGISTRY }}
|
registry: ${{ env.REGISTRY }}
|
||||||
username: ${{ github.actor }}
|
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.
|
# 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
|
- name: Extract metadata (tags, labels) for Docker
|
||||||
id: meta
|
id: meta
|
||||||
uses: docker/metadata-action@v5
|
uses: docker/metadata-action@v6
|
||||||
with:
|
with:
|
||||||
images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
|
images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
|
||||||
flavor: |
|
flavor: |
|
||||||
@@ -61,13 +61,13 @@ jobs:
|
|||||||
type=sha,format=long,suffix=${{ matrix.platform.tag-suffix }}
|
type=sha,format=long,suffix=${{ matrix.platform.tag-suffix }}
|
||||||
# This step sets up some additional capabilities to generate the provenance and sbom attestations
|
# This step sets up some additional capabilities to generate the provenance and sbom attestations
|
||||||
- name: Set up Docker Buildx
|
- name: Set up Docker Buildx
|
||||||
uses: docker/setup-buildx-action@v3
|
uses: docker/setup-buildx-action@v4
|
||||||
# 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.
|
# 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 `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.
|
# 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
|
- name: Build and push Docker image
|
||||||
id: push
|
id: push
|
||||||
uses: docker/build-push-action@v6
|
uses: docker/build-push-action@v7
|
||||||
with:
|
with:
|
||||||
file: ${{ matrix.platform.dockerfile }}
|
file: ${{ matrix.platform.dockerfile }}
|
||||||
context: .
|
context: .
|
||||||
@@ -76,4 +76,3 @@ jobs:
|
|||||||
labels: ${{ steps.meta.outputs.labels }}
|
labels: ${{ steps.meta.outputs.labels }}
|
||||||
provenance: mode=max
|
provenance: mode=max
|
||||||
sbom: true
|
sbom: true
|
||||||
|
|
||||||
@@ -12,13 +12,13 @@ jobs:
|
|||||||
|
|
||||||
steps:
|
steps:
|
||||||
- name: Checkout
|
- name: Checkout
|
||||||
uses: actions/checkout@v6
|
uses: actions/checkout@v7
|
||||||
- name: Setup Python
|
- name: Setup Python
|
||||||
uses: actions/setup-python@v6
|
uses: actions/setup-python@v7
|
||||||
with:
|
with:
|
||||||
python-version: "3.10"
|
python-version: "3.10"
|
||||||
- name: Install uv
|
- name: Install uv
|
||||||
uses: astral-sh/setup-uv@v7
|
uses: astral-sh/setup-uv@v10.0.0
|
||||||
with:
|
with:
|
||||||
enable-cache: true
|
enable-cache: true
|
||||||
- name: Install Dependencies
|
- name: Install Dependencies
|
||||||
|
|||||||
@@ -15,13 +15,13 @@ jobs:
|
|||||||
python-version: ['3.10', '3.11', '3.12', '3.13', '3.14']
|
python-version: ['3.10', '3.11', '3.12', '3.13', '3.14']
|
||||||
|
|
||||||
steps:
|
steps:
|
||||||
- uses: actions/checkout@v6
|
- uses: actions/checkout@v7
|
||||||
- name: Install uv
|
- name: Install uv
|
||||||
uses: astral-sh/setup-uv@v7
|
uses: astral-sh/setup-uv@v10.0.0
|
||||||
with:
|
with:
|
||||||
enable-cache: true
|
enable-cache: true
|
||||||
- name: Set up Python ${{ matrix.python-version }}
|
- name: Set up Python ${{ matrix.python-version }}
|
||||||
uses: actions/setup-python@v6
|
uses: actions/setup-python@v7
|
||||||
with:
|
with:
|
||||||
python-version: ${{ matrix.python-version }}
|
python-version: ${{ matrix.python-version }}
|
||||||
- name: Install Python dependencies
|
- name: Install Python dependencies
|
||||||
|
|||||||
@@ -11,10 +11,12 @@ jobs:
|
|||||||
update-ios-version:
|
update-ios-version:
|
||||||
runs-on: ubuntu-latest
|
runs-on: ubuntu-latest
|
||||||
steps:
|
steps:
|
||||||
- uses: actions/checkout@v3
|
- uses: actions/checkout@v7
|
||||||
- uses: actions/setup-python@v4
|
- uses: actions/setup-python@v7
|
||||||
|
- name: Install script dependencies
|
||||||
|
run: python -m pip install "packaging==26.3"
|
||||||
- name: Run script to fetch latest iOS releases from Apple RSS feed.
|
- name: Run script to fetch latest iOS releases from Apple RSS feed.
|
||||||
run: python3 .github/workflows/scripts/update-ios-releases.py
|
run: python .github/workflows/scripts/update-ios-releases.py
|
||||||
- name: Create Pull Request
|
- name: Create Pull Request
|
||||||
uses: peter-evans/create-pull-request@v8
|
uses: peter-evans/create-pull-request@v8
|
||||||
with:
|
with:
|
||||||
|
|||||||
@@ -82,6 +82,11 @@ Module-running `check-*` commands can load custom Python modules with
|
|||||||
[development documentation](https://docs.mvt.re/en/latest/development/) for
|
[development documentation](https://docs.mvt.re/en/latest/development/) for
|
||||||
details.
|
details.
|
||||||
|
|
||||||
|
Users can also add top-level commands to `mvt-ios` and `mvt-android` from
|
||||||
|
installed Python packages or local files and folders. See the
|
||||||
|
[custom CLI command documentation](https://docs.mvt.re/en/latest/custom_commands/)
|
||||||
|
for the plugin entry points and `--load-command` interface.
|
||||||
|
|
||||||
|
|
||||||
## License
|
## License
|
||||||
|
|
||||||
|
|||||||
@@ -57,3 +57,9 @@ If the backup is encrypted, ABE will prompt you to enter the password.
|
|||||||
Alternatively, [ab-decrypt](https://github.com/joernheissler/ab-decrypt) can be used for that purpose.
|
Alternatively, [ab-decrypt](https://github.com/joernheissler/ab-decrypt) can be used for that purpose.
|
||||||
|
|
||||||
You can then extract SMSs with MVT by passing the folder path as parameter instead of the `.ab` file: `mvt-android check-backup --output /path/to/results/ /path/to/backup/` (the path to backup given should be the folder containing the `apps` folder).
|
You can then extract SMSs with MVT by passing the folder path as parameter instead of the `.ab` file: `mvt-android check-backup --output /path/to/results/ /path/to/backup/` (the path to backup given should be the folder containing the `apps` folder).
|
||||||
|
|
||||||
|
When an output folder is specified, URLs extracted from SMS and MMS messages
|
||||||
|
are also written to `urls.json`. Each entry contains the URL, its expanded
|
||||||
|
destination when MVT resolved a shortened URL during indicator checking, the
|
||||||
|
message timestamp, and the `sms` source. The same file is created by
|
||||||
|
`check-androidqf` when its nested Android backup contains messages with URLs.
|
||||||
@@ -0,0 +1,107 @@
|
|||||||
|
# Custom CLI Commands
|
||||||
|
|
||||||
|
MVT can load additional top-level commands into `mvt-ios` and `mvt-android`.
|
||||||
|
Custom commands are different from [custom forensic modules](development.md#custom-modules):
|
||||||
|
commands add new CLI operations, while modules add analysis steps to existing
|
||||||
|
`check-*` commands.
|
||||||
|
|
||||||
|
!!! warning
|
||||||
|
|
||||||
|
Custom commands run as trusted Python code inside the MVT process. Install
|
||||||
|
or load commands only from sources you trust. MVT does not sandbox
|
||||||
|
third-party commands, and the MVT maintainers do not maintain them.
|
||||||
|
|
||||||
|
## Install a Command Package
|
||||||
|
|
||||||
|
Python packages can register a Click command or group for either MVT CLI. A
|
||||||
|
minimal package can expose this command from `my_mvt_plugin.py`:
|
||||||
|
|
||||||
|
```python
|
||||||
|
import click
|
||||||
|
|
||||||
|
|
||||||
|
@click.command()
|
||||||
|
@click.argument("path", type=click.Path(exists=True))
|
||||||
|
def summarize(path):
|
||||||
|
"""Summarize an acquisition."""
|
||||||
|
click.echo(f"Summarizing {path}")
|
||||||
|
```
|
||||||
|
|
||||||
|
Register the object in the package's `pyproject.toml`. The entry-point name is
|
||||||
|
the command users invoke:
|
||||||
|
|
||||||
|
```toml
|
||||||
|
[project.entry-points."mvt.ios.cli_plugins"]
|
||||||
|
summarize = "my_mvt_plugin:summarize"
|
||||||
|
|
||||||
|
[project.entry-points."mvt.android.cli_plugins"]
|
||||||
|
summarize = "my_mvt_plugin:summarize"
|
||||||
|
```
|
||||||
|
|
||||||
|
Use only the iOS or Android group if the command is platform-specific. After
|
||||||
|
installing the package in the same environment as MVT, it appears directly in
|
||||||
|
the appropriate CLI:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
mvt-ios summarize ./ios-backup
|
||||||
|
mvt-android summarize ./androidqf-output
|
||||||
|
```
|
||||||
|
|
||||||
|
For a `pipx` installation of MVT, inject the plugin into MVT's environment:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
pipx inject mvt my-mvt-plugin
|
||||||
|
```
|
||||||
|
|
||||||
|
When MVT is installed in an active virtual environment, install the plugin with
|
||||||
|
`pip` in that environment.
|
||||||
|
|
||||||
|
## Load a Command File
|
||||||
|
|
||||||
|
For local commands that are not packaged, create a Python file that exports one
|
||||||
|
Click command or group named `cli`:
|
||||||
|
|
||||||
|
```python
|
||||||
|
import click
|
||||||
|
|
||||||
|
|
||||||
|
@click.command("case-summary")
|
||||||
|
@click.argument("path", type=click.Path(exists=True))
|
||||||
|
def cli(path):
|
||||||
|
"""Summarize a case directory."""
|
||||||
|
click.echo(f"Summarizing {path}")
|
||||||
|
```
|
||||||
|
|
||||||
|
Pass the file before the custom command name:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
mvt-ios --load-command ./case_summary.py case-summary ./ios-backup
|
||||||
|
```
|
||||||
|
|
||||||
|
`--load-command` can be repeated and also accepts a folder. MVT loads
|
||||||
|
non-hidden top-level `*.py` files in sorted order and skips `__init__.py`.
|
||||||
|
Every loaded file must export one `cli` object.
|
||||||
|
|
||||||
|
To load a file or folder on every invocation, set the platform-specific
|
||||||
|
environment variable:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
export MVT_IOS_CUSTOM_COMMANDS=./ios_commands
|
||||||
|
export MVT_ANDROID_CUSTOM_COMMANDS=./android_commands
|
||||||
|
```
|
||||||
|
|
||||||
|
## Naming and Errors
|
||||||
|
|
||||||
|
Built-in MVT commands cannot be replaced. External command names must also be
|
||||||
|
unique; when installed packages or environment paths collide, MVT keeps the
|
||||||
|
first command and logs a warning. A collision from an explicit
|
||||||
|
`--load-command` is a usage error.
|
||||||
|
|
||||||
|
A package entry point or environment command that cannot be imported appears
|
||||||
|
as a marked broken command without preventing other MVT commands from working.
|
||||||
|
Invoke that command to see its package or file source and the underlying error.
|
||||||
|
An invalid command supplied explicitly with `--load-command` fails immediately
|
||||||
|
with a usage error.
|
||||||
|
|
||||||
|
Installed command packages use the entry-point name as the CLI command name.
|
||||||
|
The entry point must resolve to a `click.Command` or `click.Group`.
|
||||||
@@ -312,6 +312,15 @@ If indicators are provided through the command-line, they are checked against th
|
|||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
|
### `urls.json`
|
||||||
|
|
||||||
|
This JSON file collects URLs extracted from SMS, iMessage, and WhatsApp
|
||||||
|
messages. Each entry contains the original URL, its expanded destination when
|
||||||
|
MVT resolved a shortened URL during indicator checking, the message timestamp,
|
||||||
|
and its `sms` or `whatsapp` source.
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
### `sms_attachments.json`
|
### `sms_attachments.json`
|
||||||
|
|
||||||
!!! info "Availability"
|
!!! info "Availability"
|
||||||
|
|||||||
@@ -31,6 +31,7 @@ nav:
|
|||||||
- Introduction: "introduction.md"
|
- Introduction: "introduction.md"
|
||||||
- Installation: "install.md"
|
- Installation: "install.md"
|
||||||
- Command Completion: "command_completion.md"
|
- Command Completion: "command_completion.md"
|
||||||
|
- Custom CLI Commands: "custom_commands.md"
|
||||||
- Using Docker: "docker.md"
|
- Using Docker: "docker.md"
|
||||||
- MVT for iOS:
|
- MVT for iOS:
|
||||||
- iOS Forensic Methodology: "ios/methodology.md"
|
- iOS Forensic Methodology: "ios/methodology.md"
|
||||||
|
|||||||
+13
-13
@@ -17,25 +17,25 @@ classifiers = [
|
|||||||
"Programming Language :: Python",
|
"Programming Language :: Python",
|
||||||
]
|
]
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"click==8.3.3",
|
"click==8.4.2",
|
||||||
"rich==14.3.3",
|
"rich==15.0.0",
|
||||||
"tld==0.13.2",
|
"tld==0.13.2",
|
||||||
"requests==2.33.1",
|
"requests==2.34.2",
|
||||||
"simplejson==3.20.2",
|
"simplejson==4.1.1",
|
||||||
"packaging==26.2",
|
"packaging==26.3",
|
||||||
"appdirs==1.4.4",
|
"appdirs==1.4.4",
|
||||||
"iOSbackup==0.9.925",
|
"iOSbackup==0.9.925",
|
||||||
"adb-shell[usb]==0.4.4",
|
"adb-shell[usb]==0.4.4",
|
||||||
"libusb1==3.3.1",
|
"libusb1==3.4.0",
|
||||||
"cryptography==47.0.0",
|
"cryptography==50.0.0",
|
||||||
"PyYAML>=6.0.2",
|
"PyYAML>=6.0.2",
|
||||||
"pyahocorasick==2.3.1",
|
"pyahocorasick==2.3.1",
|
||||||
"betterproto2==0.9.1",
|
"betterproto2==0.10.0",
|
||||||
"pydantic==2.13.3",
|
"pydantic==2.13.4",
|
||||||
"pydantic-settings==2.14.0",
|
"pydantic-settings==2.15.0",
|
||||||
"NSKeyedUnArchiver==1.5.2",
|
"NSKeyedUnArchiver==1.5.2",
|
||||||
"python-dateutil==2.9.0.post0",
|
"python-dateutil==2.9.0.post0",
|
||||||
"tzdata==2026.2",
|
"tzdata==2026.3",
|
||||||
]
|
]
|
||||||
requires-python = ">= 3.10"
|
requires-python = ">= 3.10"
|
||||||
|
|
||||||
@@ -65,9 +65,9 @@ dev = [
|
|||||||
docs = [
|
docs = [
|
||||||
"mkdocs==1.6.1",
|
"mkdocs==1.6.1",
|
||||||
"mkdocs-autorefs==1.4.4",
|
"mkdocs-autorefs==1.4.4",
|
||||||
"mkdocs-material==9.7.6",
|
"mkdocs-material==9.7.7",
|
||||||
"mkdocs-material-extensions==1.3.1",
|
"mkdocs-material-extensions==1.3.1",
|
||||||
"mkdocstrings==1.0.4",
|
"mkdocstrings==1.0.6",
|
||||||
]
|
]
|
||||||
|
|
||||||
[build-system]
|
[build-system]
|
||||||
|
|||||||
@@ -34,6 +34,7 @@ TOMBSTONE_TEXT_KEY_MAPPINGS = {
|
|||||||
"signal": "signal_info",
|
"signal": "signal_info",
|
||||||
"code": "code",
|
"code": "code",
|
||||||
"Cause": "cause",
|
"Cause": "cause",
|
||||||
|
"Abort message": "abort_message",
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
@@ -67,6 +68,8 @@ class TombstoneCrashResult(pydantic.BaseModel):
|
|||||||
uid: int
|
uid: int
|
||||||
signal_info: SignalInfo
|
signal_info: SignalInfo
|
||||||
cause: Optional[str] = None
|
cause: Optional[str] = None
|
||||||
|
causes: Optional[List[dict]] = None
|
||||||
|
abort_message: Optional[str] = None
|
||||||
extra: Optional[str] = None
|
extra: Optional[str] = None
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -8,6 +8,12 @@ from zipfile import BadZipFile
|
|||||||
|
|
||||||
import click
|
import click
|
||||||
|
|
||||||
|
from mvt.common.cli_plugins import (
|
||||||
|
ANDROID_CLI_PLUGIN_GROUP,
|
||||||
|
MVT_ANDROID_CUSTOM_COMMANDS_ENV,
|
||||||
|
load_cli_commands_option,
|
||||||
|
register_cli_plugins,
|
||||||
|
)
|
||||||
from mvt.common.cmd_check_iocs import CmdCheckIOCS
|
from mvt.common.cmd_check_iocs import CmdCheckIOCS
|
||||||
from mvt.common.completion import (
|
from mvt.common.completion import (
|
||||||
SUPPORTED_SHELLS,
|
SUPPORTED_SHELLS,
|
||||||
@@ -82,6 +88,7 @@ def _load_custom_modules(load_module):
|
|||||||
# Main
|
# Main
|
||||||
# ==============================================================================
|
# ==============================================================================
|
||||||
@click.group(invoke_without_command=False)
|
@click.group(invoke_without_command=False)
|
||||||
|
@load_cli_commands_option
|
||||||
@click.option(
|
@click.option(
|
||||||
"--disable-update-check", is_flag=True, help=HELP_MSG_DISABLE_UPDATE_CHECK
|
"--disable-update-check", is_flag=True, help=HELP_MSG_DISABLE_UPDATE_CHECK
|
||||||
)
|
)
|
||||||
@@ -505,3 +512,10 @@ def check_iocs(ctx, iocs, list_modules, module, load_module, folder):
|
|||||||
def download_indicators():
|
def download_indicators():
|
||||||
ioc_updates = IndicatorsUpdates()
|
ioc_updates = IndicatorsUpdates()
|
||||||
ioc_updates.update()
|
ioc_updates.update()
|
||||||
|
|
||||||
|
|
||||||
|
register_cli_plugins(
|
||||||
|
cli,
|
||||||
|
entry_point_group=ANDROID_CLI_PLUGIN_GROUP,
|
||||||
|
environment_variable=MVT_ANDROID_CUSTOM_COMMANDS_ENV,
|
||||||
|
)
|
||||||
@@ -262,6 +262,7 @@ class CmdAndroidCheckAndroidQF(Command):
|
|||||||
cmd.run()
|
cmd.run()
|
||||||
|
|
||||||
self.timeline.extend(cmd.timeline)
|
self.timeline.extend(cmd.timeline)
|
||||||
|
self.url_results.extend(cmd.url_results)
|
||||||
self.alertstore.extend(cmd.alertstore.alerts)
|
self.alertstore.extend(cmd.alertstore.alerts)
|
||||||
finally:
|
finally:
|
||||||
if bugreport:
|
if bugreport:
|
||||||
@@ -299,6 +300,7 @@ class CmdAndroidCheckAndroidQF(Command):
|
|||||||
cmd.run()
|
cmd.run()
|
||||||
|
|
||||||
self.timeline.extend(cmd.timeline)
|
self.timeline.extend(cmd.timeline)
|
||||||
|
self.url_results.extend(cmd.url_results)
|
||||||
self.alertstore.extend(cmd.alertstore.alerts)
|
self.alertstore.extend(cmd.alertstore.alerts)
|
||||||
return True
|
return True
|
||||||
|
|
||||||
@@ -378,6 +380,7 @@ class CmdAndroidCheckAndroidQF(Command):
|
|||||||
cmd.run()
|
cmd.run()
|
||||||
|
|
||||||
self.timeline.extend(cmd.timeline)
|
self.timeline.extend(cmd.timeline)
|
||||||
|
self.url_results.extend(cmd.url_results)
|
||||||
self.alertstore.extend(cmd.alertstore.alerts)
|
self.alertstore.extend(cmd.alertstore.alerts)
|
||||||
return True
|
return True
|
||||||
|
|
||||||
|
|||||||
@@ -57,6 +57,11 @@ class SMS(BackupModule):
|
|||||||
ioc_match.message, "", message, matched_indicator=ioc_match.ioc
|
ioc_match.message, "", message, matched_indicator=ioc_match.ioc
|
||||||
)
|
)
|
||||||
|
|
||||||
|
def collect_url_results(self) -> None:
|
||||||
|
for message in self.results:
|
||||||
|
for url in message.get("links", []):
|
||||||
|
self.add_url_result(url, message.get("isodate"), "sms")
|
||||||
|
|
||||||
def run(self) -> None:
|
def run(self) -> None:
|
||||||
sms_path = "apps/com.android.providers.telephony/d_f/*_sms_backup"
|
sms_path = "apps/com.android.providers.telephony/d_f/*_sms_backup"
|
||||||
for file in self._get_files_by_pattern(sms_path):
|
for file in self._get_files_by_pattern(sms_path):
|
||||||
|
|||||||
@@ -0,0 +1,297 @@
|
|||||||
|
# Mobile Verification Toolkit (MVT)
|
||||||
|
# Copyright (c) 2021-2026 The MVT Authors.
|
||||||
|
# Use of this software is governed by the MVT License 1.1 that can be found at
|
||||||
|
# https://license.mvt.re/1.1/
|
||||||
|
|
||||||
|
import hashlib
|
||||||
|
import importlib.metadata
|
||||||
|
import importlib.util
|
||||||
|
import logging
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
from pathlib import Path
|
||||||
|
from types import ModuleType
|
||||||
|
from typing import Iterable
|
||||||
|
|
||||||
|
import click
|
||||||
|
|
||||||
|
IOS_CLI_PLUGIN_GROUP = "mvt.ios.cli_plugins"
|
||||||
|
ANDROID_CLI_PLUGIN_GROUP = "mvt.android.cli_plugins"
|
||||||
|
MVT_IOS_CUSTOM_COMMANDS_ENV = "MVT_IOS_CUSTOM_COMMANDS"
|
||||||
|
MVT_ANDROID_CUSTOM_COMMANDS_ENV = "MVT_ANDROID_CUSTOM_COMMANDS"
|
||||||
|
|
||||||
|
log = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
class CustomCommandLoadError(Exception):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class BrokenPluginCommand(click.Command):
|
||||||
|
"""A placeholder for an installed or configured command that failed to load."""
|
||||||
|
|
||||||
|
def __init__(self, name: str, source: str, exception: BaseException):
|
||||||
|
super().__init__(
|
||||||
|
name,
|
||||||
|
help=(
|
||||||
|
f"Unable to load external command from {source}.\n\n"
|
||||||
|
f"{type(exception).__name__}: {exception}"
|
||||||
|
),
|
||||||
|
short_help="Warning: external command could not be loaded.",
|
||||||
|
)
|
||||||
|
self.source = source
|
||||||
|
self.exception = exception
|
||||||
|
|
||||||
|
def parse_args(self, ctx: click.Context, args: list[str]) -> list[str]:
|
||||||
|
return args
|
||||||
|
|
||||||
|
def invoke(self, ctx: click.Context) -> None:
|
||||||
|
raise click.ClickException(
|
||||||
|
f"Unable to load external command '{self.name}' from {self.source}: "
|
||||||
|
f"{type(self.exception).__name__}: {self.exception}"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _module_name_for_path(path: Path) -> str:
|
||||||
|
digest = hashlib.sha256(str(path).encode("utf-8")).hexdigest()[:16]
|
||||||
|
return f"_mvt_custom_command_{path.stem}_{digest}"
|
||||||
|
|
||||||
|
|
||||||
|
def _iter_command_files(path: Path) -> Iterable[Path]:
|
||||||
|
if path.is_file():
|
||||||
|
if path.suffix != ".py":
|
||||||
|
raise CustomCommandLoadError(
|
||||||
|
f"Custom command file is not a Python file: {path}"
|
||||||
|
)
|
||||||
|
yield path
|
||||||
|
return
|
||||||
|
|
||||||
|
if path.is_dir():
|
||||||
|
for child in sorted(path.iterdir()):
|
||||||
|
if child.name.startswith(".") or child.name == "__init__.py":
|
||||||
|
continue
|
||||||
|
if child.is_file() and child.suffix == ".py":
|
||||||
|
yield child
|
||||||
|
return
|
||||||
|
|
||||||
|
raise CustomCommandLoadError(f"Custom command path does not exist: {path}")
|
||||||
|
|
||||||
|
|
||||||
|
def _load_python_file(path: Path) -> ModuleType:
|
||||||
|
module_name = _module_name_for_path(path)
|
||||||
|
spec = importlib.util.spec_from_file_location(module_name, path)
|
||||||
|
if spec is None or spec.loader is None:
|
||||||
|
raise CustomCommandLoadError(f"Unable to load custom command file: {path}")
|
||||||
|
|
||||||
|
module = importlib.util.module_from_spec(spec)
|
||||||
|
sys.modules[module_name] = module
|
||||||
|
try:
|
||||||
|
spec.loader.exec_module(module)
|
||||||
|
except (Exception, SystemExit) as exc:
|
||||||
|
raise CustomCommandLoadError(
|
||||||
|
f"Unable to import custom command {path}: {exc}"
|
||||||
|
) from exc
|
||||||
|
|
||||||
|
return module
|
||||||
|
|
||||||
|
|
||||||
|
def load_cli_command_file(path: Path) -> click.Command:
|
||||||
|
module = _load_python_file(path)
|
||||||
|
command = getattr(module, "cli", None)
|
||||||
|
if not isinstance(command, click.Command):
|
||||||
|
raise CustomCommandLoadError(
|
||||||
|
f"Custom command {path} must export a Click command or group named 'cli'"
|
||||||
|
)
|
||||||
|
if command.name is None:
|
||||||
|
raise CustomCommandLoadError(f"Custom command {path} has no command name")
|
||||||
|
return command
|
||||||
|
|
||||||
|
|
||||||
|
def _register_command(
|
||||||
|
group: click.Group,
|
||||||
|
command: click.Command,
|
||||||
|
*,
|
||||||
|
name: str,
|
||||||
|
source: str,
|
||||||
|
collision_is_error: bool,
|
||||||
|
) -> bool:
|
||||||
|
if name in group.commands:
|
||||||
|
registered_sources = getattr(group, "_mvt_external_command_sources", {})
|
||||||
|
if registered_sources.get(name) == source:
|
||||||
|
return False
|
||||||
|
message = (
|
||||||
|
f"Unable to register external command '{name}' from {source}: "
|
||||||
|
"the command name is already registered"
|
||||||
|
)
|
||||||
|
if collision_is_error:
|
||||||
|
raise CustomCommandLoadError(message)
|
||||||
|
log.warning(message)
|
||||||
|
return False
|
||||||
|
|
||||||
|
group.add_command(command, name=name)
|
||||||
|
registered_sources = getattr(group, "_mvt_external_command_sources", {})
|
||||||
|
registered_sources[name] = source
|
||||||
|
setattr(group, "_mvt_external_command_sources", registered_sources)
|
||||||
|
return True
|
||||||
|
|
||||||
|
|
||||||
|
def register_cli_commands_from_path(
|
||||||
|
group: click.Group,
|
||||||
|
path: str | Path,
|
||||||
|
*,
|
||||||
|
collision_is_error: bool = False,
|
||||||
|
failures_are_errors: bool = False,
|
||||||
|
) -> list[str]:
|
||||||
|
resolved_path = Path(path).expanduser().resolve()
|
||||||
|
registered: list[str] = []
|
||||||
|
|
||||||
|
try:
|
||||||
|
command_files = list(_iter_command_files(resolved_path))
|
||||||
|
except CustomCommandLoadError as exc:
|
||||||
|
if failures_are_errors:
|
||||||
|
raise
|
||||||
|
name = resolved_path.stem.replace("_", "-") or "custom-command"
|
||||||
|
if _register_command(
|
||||||
|
group,
|
||||||
|
BrokenPluginCommand(name, str(resolved_path), exc),
|
||||||
|
name=name,
|
||||||
|
source=str(resolved_path),
|
||||||
|
collision_is_error=collision_is_error,
|
||||||
|
):
|
||||||
|
registered.append(name)
|
||||||
|
return registered
|
||||||
|
|
||||||
|
for command_file in command_files:
|
||||||
|
try:
|
||||||
|
command = load_cli_command_file(command_file)
|
||||||
|
except CustomCommandLoadError as exc:
|
||||||
|
if failures_are_errors:
|
||||||
|
raise
|
||||||
|
name = command_file.stem.replace("_", "-")
|
||||||
|
command = BrokenPluginCommand(name, str(command_file), exc)
|
||||||
|
|
||||||
|
command_name = command.name
|
||||||
|
if command_name is None:
|
||||||
|
raise CustomCommandLoadError(
|
||||||
|
f"Custom command {command_file} has no command name"
|
||||||
|
)
|
||||||
|
if _register_command(
|
||||||
|
group,
|
||||||
|
command,
|
||||||
|
name=command_name,
|
||||||
|
source=str(command_file),
|
||||||
|
collision_is_error=collision_is_error,
|
||||||
|
):
|
||||||
|
registered.append(command_name)
|
||||||
|
|
||||||
|
return registered
|
||||||
|
|
||||||
|
|
||||||
|
def _entry_point_source(entry_point: importlib.metadata.EntryPoint) -> str:
|
||||||
|
try:
|
||||||
|
distribution = getattr(entry_point, "dist", None)
|
||||||
|
if distribution is None:
|
||||||
|
return f"entry point {entry_point.value}"
|
||||||
|
|
||||||
|
name = distribution.metadata.get("Name", "unknown distribution")
|
||||||
|
version = getattr(distribution, "version", None)
|
||||||
|
if version:
|
||||||
|
return f"{name} {version} ({entry_point.value})"
|
||||||
|
return f"{name} ({entry_point.value})"
|
||||||
|
except Exception:
|
||||||
|
return f"entry point {entry_point.value}"
|
||||||
|
|
||||||
|
|
||||||
|
def register_installed_cli_commands(
|
||||||
|
group: click.Group,
|
||||||
|
entry_point_group: str,
|
||||||
|
) -> list[str]:
|
||||||
|
try:
|
||||||
|
entry_points = importlib.metadata.entry_points(group=entry_point_group)
|
||||||
|
except Exception as exc:
|
||||||
|
log.warning(
|
||||||
|
"Unable to discover external commands in entry-point group %s: %s",
|
||||||
|
entry_point_group,
|
||||||
|
exc,
|
||||||
|
)
|
||||||
|
return []
|
||||||
|
ordered_entry_points = sorted(
|
||||||
|
entry_points,
|
||||||
|
key=lambda entry_point: (
|
||||||
|
entry_point.name,
|
||||||
|
_entry_point_source(entry_point),
|
||||||
|
entry_point.value,
|
||||||
|
),
|
||||||
|
)
|
||||||
|
registered: list[str] = []
|
||||||
|
|
||||||
|
for entry_point in ordered_entry_points:
|
||||||
|
source = _entry_point_source(entry_point)
|
||||||
|
try:
|
||||||
|
command = entry_point.load()
|
||||||
|
if not isinstance(command, click.Command):
|
||||||
|
raise TypeError(
|
||||||
|
f"entry point must resolve to a Click command or group, "
|
||||||
|
f"not {type(command).__name__}"
|
||||||
|
)
|
||||||
|
except (Exception, SystemExit) as exc:
|
||||||
|
command = BrokenPluginCommand(entry_point.name, source, exc)
|
||||||
|
|
||||||
|
if _register_command(
|
||||||
|
group,
|
||||||
|
command,
|
||||||
|
name=entry_point.name,
|
||||||
|
source=source,
|
||||||
|
collision_is_error=False,
|
||||||
|
):
|
||||||
|
registered.append(entry_point.name)
|
||||||
|
|
||||||
|
return registered
|
||||||
|
|
||||||
|
|
||||||
|
def register_cli_plugins(
|
||||||
|
group: click.Group,
|
||||||
|
*,
|
||||||
|
entry_point_group: str,
|
||||||
|
environment_variable: str,
|
||||||
|
) -> None:
|
||||||
|
environment_path = os.environ.get(environment_variable)
|
||||||
|
if environment_path:
|
||||||
|
register_cli_commands_from_path(group, environment_path)
|
||||||
|
register_installed_cli_commands(group, entry_point_group)
|
||||||
|
|
||||||
|
|
||||||
|
def _load_command_option_callback(
|
||||||
|
ctx: click.Context,
|
||||||
|
param: click.Parameter,
|
||||||
|
paths: tuple[str, ...],
|
||||||
|
) -> tuple[str, ...]:
|
||||||
|
if not isinstance(ctx.command, click.Group):
|
||||||
|
raise click.ClickException("--load-command requires a Click command group")
|
||||||
|
|
||||||
|
for path in paths:
|
||||||
|
try:
|
||||||
|
register_cli_commands_from_path(
|
||||||
|
ctx.command,
|
||||||
|
path,
|
||||||
|
collision_is_error=True,
|
||||||
|
failures_are_errors=True,
|
||||||
|
)
|
||||||
|
except CustomCommandLoadError as exc:
|
||||||
|
raise click.BadParameter(str(exc), ctx=ctx, param=param) from exc
|
||||||
|
return paths
|
||||||
|
|
||||||
|
|
||||||
|
load_cli_commands_option = click.option(
|
||||||
|
"--load-command",
|
||||||
|
"load_commands",
|
||||||
|
type=click.Path(exists=True),
|
||||||
|
multiple=True,
|
||||||
|
expose_value=False,
|
||||||
|
is_eager=True,
|
||||||
|
callback=_load_command_option_callback,
|
||||||
|
help=(
|
||||||
|
"Load a custom CLI command from a Python file or folder "
|
||||||
|
"(can be invoked multiple times)"
|
||||||
|
),
|
||||||
|
)
|
||||||
@@ -20,7 +20,7 @@ from .config import settings
|
|||||||
from .indicators import Indicators
|
from .indicators import Indicators
|
||||||
from .module import EncryptedBackupError, MVTModule, run_module, save_timeline
|
from .module import EncryptedBackupError, MVTModule, run_module, save_timeline
|
||||||
from .module_loader import module_supports_command
|
from .module_loader import module_supports_command
|
||||||
from .module_types import ModuleTimeline
|
from .module_types import ModuleTimeline, URLResult
|
||||||
from .utils import (
|
from .utils import (
|
||||||
CustomJSONEncoder,
|
CustomJSONEncoder,
|
||||||
convert_datetime_to_iso,
|
convert_datetime_to_iso,
|
||||||
@@ -73,6 +73,7 @@ class Command:
|
|||||||
self.hashes = hashes
|
self.hashes = hashes
|
||||||
self.hash_values: list[dict[str, Any]] = []
|
self.hash_values: list[dict[str, Any]] = []
|
||||||
self.timeline: ModuleTimeline = []
|
self.timeline: ModuleTimeline = []
|
||||||
|
self.url_results: list[URLResult] = []
|
||||||
|
|
||||||
# Load IOCs
|
# Load IOCs
|
||||||
self._create_storage()
|
self._create_storage()
|
||||||
@@ -150,6 +151,14 @@ class Command:
|
|||||||
with open(alerts_path, "w+", encoding="utf-8") as handle:
|
with open(alerts_path, "w+", encoding="utf-8") as handle:
|
||||||
json.dump(alerts, handle, indent=4, cls=CustomJSONEncoder)
|
json.dump(alerts, handle, indent=4, cls=CustomJSONEncoder)
|
||||||
|
|
||||||
|
def _store_urls(self) -> None:
|
||||||
|
if not self.results_path or not self.url_results:
|
||||||
|
return
|
||||||
|
|
||||||
|
urls_path = os.path.join(self.results_path, "urls.json")
|
||||||
|
with open(urls_path, "w", encoding="utf-8") as handle:
|
||||||
|
json.dump(self.url_results, handle, indent=4, cls=CustomJSONEncoder)
|
||||||
|
|
||||||
def _store_alerts_timeline(self) -> None:
|
def _store_alerts_timeline(self) -> None:
|
||||||
if not self.results_path:
|
if not self.results_path:
|
||||||
return
|
return
|
||||||
@@ -396,6 +405,7 @@ class Command:
|
|||||||
self.executed.append(m)
|
self.executed.append(m)
|
||||||
executed_by_type[module] = m
|
executed_by_type[module] = m
|
||||||
self.timeline.extend(m.timeline)
|
self.timeline.extend(m.timeline)
|
||||||
|
self.url_results.extend(m.url_results)
|
||||||
self.alertstore.extend(m.alertstore.alerts)
|
self.alertstore.extend(m.alertstore.alerts)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
@@ -410,4 +420,5 @@ class Command:
|
|||||||
self._store_timeline()
|
self._store_timeline()
|
||||||
self._store_alerts_timeline()
|
self._store_alerts_timeline()
|
||||||
self._store_alerts()
|
self._store_alerts()
|
||||||
|
self._store_urls()
|
||||||
self._store_info()
|
self._store_info()
|
||||||
@@ -49,6 +49,7 @@ class Indicators:
|
|||||||
self.log = log
|
self.log = log
|
||||||
self.ioc_collections: List[Dict[str, Any]] = []
|
self.ioc_collections: List[Dict[str, Any]] = []
|
||||||
self.total_ioc_count = 0
|
self.total_ioc_count = 0
|
||||||
|
self.resolved_urls: Dict[str, str] = {}
|
||||||
|
|
||||||
def _load_downloaded_indicators(self) -> None:
|
def _load_downloaded_indicators(self) -> None:
|
||||||
if not os.path.isdir(MVT_INDICATORS_FOLDER):
|
if not os.path.isdir(MVT_INDICATORS_FOLDER):
|
||||||
@@ -439,9 +440,14 @@ class Indicators:
|
|||||||
orig_url.url,
|
orig_url.url,
|
||||||
dest_url.url,
|
dest_url.url,
|
||||||
)
|
)
|
||||||
return self.check_url(dest_url.url)
|
match = self.check_url(dest_url.url)
|
||||||
|
self.resolved_urls[url] = self.resolved_urls.get(
|
||||||
|
dest_url.url, dest_url.url
|
||||||
|
)
|
||||||
|
return match
|
||||||
|
|
||||||
final_url = dest_url
|
final_url = dest_url
|
||||||
|
self.resolved_urls[url] = final_url.url
|
||||||
else:
|
else:
|
||||||
# If it's not shortened, we just use the original URL object.
|
# If it's not shortened, we just use the original URL object.
|
||||||
final_url = orig_url
|
final_url = orig_url
|
||||||
@@ -482,6 +488,13 @@ class Indicators:
|
|||||||
|
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
def get_expanded_url(self, url: str) -> Optional[str]:
|
||||||
|
"""Return the final URL recorded while checking a shortened URL."""
|
||||||
|
expanded_url = self.resolved_urls.get(url)
|
||||||
|
if expanded_url and expanded_url != url:
|
||||||
|
return expanded_url
|
||||||
|
return None
|
||||||
|
|
||||||
def check_urls(self, urls: list) -> Optional[IndicatorMatch]:
|
def check_urls(self, urls: list) -> Optional[IndicatorMatch]:
|
||||||
"""Check a list of URLs against the provided list of domain indicators.
|
"""Check a list of URLs against the provided list of domain indicators.
|
||||||
|
|
||||||
|
|||||||
@@ -18,6 +18,7 @@ from .module_types import (
|
|||||||
ModuleResults,
|
ModuleResults,
|
||||||
ModuleSerializedResult,
|
ModuleSerializedResult,
|
||||||
ModuleTimeline,
|
ModuleTimeline,
|
||||||
|
URLResult,
|
||||||
)
|
)
|
||||||
from .utils import CustomJSONEncoder, exec_or_profile
|
from .utils import CustomJSONEncoder, exec_or_profile
|
||||||
|
|
||||||
@@ -82,6 +83,7 @@ class MVTModule:
|
|||||||
|
|
||||||
self.results: ModuleResults = results if results is not None else []
|
self.results: ModuleResults = results if results is not None else []
|
||||||
self.timeline: ModuleTimeline = []
|
self.timeline: ModuleTimeline = []
|
||||||
|
self.url_results: list[URLResult] = []
|
||||||
self.dependency_modules: Dict[type["MVTModule"], "MVTModule"] = {}
|
self.dependency_modules: Dict[type["MVTModule"], "MVTModule"] = {}
|
||||||
|
|
||||||
def get_dependency_results(
|
def get_dependency_results(
|
||||||
@@ -110,6 +112,23 @@ class MVTModule:
|
|||||||
def check_indicators(self) -> None:
|
def check_indicators(self) -> None:
|
||||||
raise NotImplementedError
|
raise NotImplementedError
|
||||||
|
|
||||||
|
def collect_url_results(self) -> None:
|
||||||
|
"""Collect URL records exposed by this module."""
|
||||||
|
|
||||||
|
def add_url_result(self, url: str, timestamp: Optional[str], source: str) -> None:
|
||||||
|
expanded_url = None
|
||||||
|
if self.indicators:
|
||||||
|
expanded_url = self.indicators.get_expanded_url(url)
|
||||||
|
|
||||||
|
self.url_results.append(
|
||||||
|
{
|
||||||
|
"url": url,
|
||||||
|
"expanded_url": expanded_url,
|
||||||
|
"timestamp": timestamp,
|
||||||
|
"source": source,
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
def save_to_json(self) -> None:
|
def save_to_json(self) -> None:
|
||||||
if not self.results_path:
|
if not self.results_path:
|
||||||
return
|
return
|
||||||
@@ -249,6 +268,15 @@ def run_module(module: MVTModule) -> None:
|
|||||||
"The %s module produced no detections!", module.__class__.__name__
|
"The %s module produced no detections!", module.__class__.__name__
|
||||||
)
|
)
|
||||||
|
|
||||||
|
try:
|
||||||
|
module.collect_url_results()
|
||||||
|
except Exception as exc:
|
||||||
|
module.log.exception(
|
||||||
|
"Error when collecting URLs from module %s: %s",
|
||||||
|
module.__class__.__name__,
|
||||||
|
exc,
|
||||||
|
)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
module.to_timeline()
|
module.to_timeline()
|
||||||
except NotImplementedError:
|
except NotImplementedError:
|
||||||
|
|||||||
@@ -4,7 +4,7 @@
|
|||||||
# https://license.mvt.re/1.1/
|
# https://license.mvt.re/1.1/
|
||||||
|
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from typing import Any, Dict, List, Union
|
from typing import Any, Dict, List, Optional, TypedDict, Union
|
||||||
|
|
||||||
|
|
||||||
# ModuleAtomicResult is a flexible dictionary that can contain any data.
|
# ModuleAtomicResult is a flexible dictionary that can contain any data.
|
||||||
@@ -22,6 +22,13 @@ ModuleAtomicResult = Dict[str, Any]
|
|||||||
ModuleResults = Any
|
ModuleResults = Any
|
||||||
|
|
||||||
|
|
||||||
|
class URLResult(TypedDict):
|
||||||
|
url: str
|
||||||
|
expanded_url: Optional[str]
|
||||||
|
timestamp: Optional[str]
|
||||||
|
source: str
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class ModuleAtomicTimeline:
|
class ModuleAtomicTimeline:
|
||||||
timestamp: str
|
timestamp: str
|
||||||
|
|||||||
@@ -8,6 +8,12 @@ import logging
|
|||||||
import os
|
import os
|
||||||
|
|
||||||
import click
|
import click
|
||||||
|
from mvt.common.cli_plugins import (
|
||||||
|
IOS_CLI_PLUGIN_GROUP,
|
||||||
|
MVT_IOS_CUSTOM_COMMANDS_ENV,
|
||||||
|
load_cli_commands_option,
|
||||||
|
register_cli_plugins,
|
||||||
|
)
|
||||||
from mvt.common.cmd_check_iocs import CmdCheckIOCS
|
from mvt.common.cmd_check_iocs import CmdCheckIOCS
|
||||||
from mvt.common.completion import (
|
from mvt.common.completion import (
|
||||||
SUPPORTED_SHELLS,
|
SUPPORTED_SHELLS,
|
||||||
@@ -86,6 +92,7 @@ def _load_custom_modules(load_module):
|
|||||||
# Main
|
# Main
|
||||||
# ==============================================================================
|
# ==============================================================================
|
||||||
@click.group(invoke_without_command=False)
|
@click.group(invoke_without_command=False)
|
||||||
|
@load_cli_commands_option
|
||||||
@click.option(
|
@click.option(
|
||||||
"--disable-update-check", is_flag=True, help=HELP_MSG_DISABLE_UPDATE_CHECK
|
"--disable-update-check", is_flag=True, help=HELP_MSG_DISABLE_UPDATE_CHECK
|
||||||
)
|
)
|
||||||
@@ -523,3 +530,10 @@ def check_iocs(ctx, iocs, list_modules, module, load_module, folder):
|
|||||||
def download_iocs():
|
def download_iocs():
|
||||||
ioc_updates = IndicatorsUpdates()
|
ioc_updates = IndicatorsUpdates()
|
||||||
ioc_updates.update()
|
ioc_updates.update()
|
||||||
|
|
||||||
|
|
||||||
|
register_cli_plugins(
|
||||||
|
cli,
|
||||||
|
entry_point_group=IOS_CLI_PLUGIN_GROUP,
|
||||||
|
environment_variable=MVT_IOS_CUSTOM_COMMANDS_ENV,
|
||||||
|
)
|
||||||
@@ -6,8 +6,10 @@
|
|||||||
import json
|
import json
|
||||||
import logging
|
import logging
|
||||||
import os
|
import os
|
||||||
|
import shutil
|
||||||
import tarfile
|
import tarfile
|
||||||
from pathlib import Path
|
from pathlib import Path, PurePosixPath
|
||||||
|
from tempfile import TemporaryDirectory
|
||||||
from typing import Any, Optional
|
from typing import Any, Optional
|
||||||
|
|
||||||
from mvt.common.command import Command
|
from mvt.common.command import Command
|
||||||
@@ -54,6 +56,8 @@ class CmdIOSCheckSysdiagnose(Command):
|
|||||||
self.sysdiagnose_archive: Optional[tarfile.TarFile] = None
|
self.sysdiagnose_archive: Optional[tarfile.TarFile] = None
|
||||||
self.sysdiagnose_files: list[str] = []
|
self.sysdiagnose_files: list[str] = []
|
||||||
self.ips_files: list[dict[str, Any]] = []
|
self.ips_files: list[dict[str, Any]] = []
|
||||||
|
self.temp_sysdiagnose_dir: Optional[TemporaryDirectory[str]] = None
|
||||||
|
self.extracted_sysdiagnose_path: Optional[str] = None
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def _parse_bugtype_header(data: bytes) -> Optional[int]:
|
def _parse_bugtype_header(data: bytes) -> Optional[int]:
|
||||||
@@ -91,21 +95,68 @@ class CmdIOSCheckSysdiagnose(Command):
|
|||||||
self.log.info("Parsing sysdiagnose archive. This might take a while...")
|
self.log.info("Parsing sysdiagnose archive. This might take a while...")
|
||||||
self.sysdiagnose_format = "tar"
|
self.sysdiagnose_format = "tar"
|
||||||
self.sysdiagnose_archive = tarfile.open(self.target_path, "r:gz")
|
self.sysdiagnose_archive = tarfile.open(self.target_path, "r:gz")
|
||||||
for member in self.sysdiagnose_archive:
|
self._extract_sysdiagnose_archive()
|
||||||
self.sysdiagnose_files.append(member.name)
|
|
||||||
if member.isfile() and member.name.endswith(".ips"):
|
def _extract_sysdiagnose_archive(self) -> None:
|
||||||
archive_handle = self.sysdiagnose_archive.extractfile(member)
|
archive = self.sysdiagnose_archive
|
||||||
if archive_handle is not None:
|
if archive is None:
|
||||||
with archive_handle:
|
raise RuntimeError("Sysdiagnose archive has not been initialized")
|
||||||
self._add_ips_file(member.name, archive_handle.read())
|
|
||||||
|
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())
|
||||||
|
|
||||||
def module_init(self, module) -> None:
|
def module_init(self, module) -> None:
|
||||||
module.ips_files = self.ips_files
|
module.ips_files = self.ips_files
|
||||||
if self.sysdiagnose_format == "tar":
|
if self.sysdiagnose_format == "tar":
|
||||||
if self.sysdiagnose_archive is None:
|
if self.extracted_sysdiagnose_path is None:
|
||||||
raise RuntimeError("Sysdiagnose archive has not been initialized")
|
raise RuntimeError("Sysdiagnose archive has not been extracted")
|
||||||
module.from_sysdiagnose_tar(
|
module.from_sysdiagnose_folder(
|
||||||
self.sysdiagnose_archive, self.sysdiagnose_files
|
self.extracted_sysdiagnose_path, self.sysdiagnose_files
|
||||||
)
|
)
|
||||||
return
|
return
|
||||||
if self.sysdiagnose_format == "dir" and self.target_path:
|
if self.sysdiagnose_format == "dir" and self.target_path:
|
||||||
@@ -117,3 +168,6 @@ class CmdIOSCheckSysdiagnose(Command):
|
|||||||
if self.sysdiagnose_archive is not None:
|
if self.sysdiagnose_archive is not None:
|
||||||
self.sysdiagnose_archive.close()
|
self.sysdiagnose_archive.close()
|
||||||
self.sysdiagnose_archive = None
|
self.sysdiagnose_archive = None
|
||||||
|
if self.temp_sysdiagnose_dir is not None:
|
||||||
|
self.temp_sysdiagnose_dir.cleanup()
|
||||||
|
self.temp_sysdiagnose_dir = None
|
||||||
@@ -1216,6 +1216,10 @@
|
|||||||
"version": "18.7.9",
|
"version": "18.7.9",
|
||||||
"build": "22H355"
|
"build": "22H355"
|
||||||
},
|
},
|
||||||
|
{
|
||||||
|
"version": "18.7.10",
|
||||||
|
"build": "22H373"
|
||||||
|
},
|
||||||
{
|
{
|
||||||
"version": "26",
|
"version": "26",
|
||||||
"build": "23A341"
|
"build": "23A341"
|
||||||
@@ -1255,5 +1259,9 @@
|
|||||||
{
|
{
|
||||||
"version": "26.5",
|
"version": "26.5",
|
||||||
"build": "23F77"
|
"build": "23F77"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"version": "26.6.1",
|
||||||
|
"build": "23G82"
|
||||||
}
|
}
|
||||||
]
|
]
|
||||||
@@ -73,7 +73,9 @@ class Calls(IOSExtraction):
|
|||||||
"duration": row[1],
|
"duration": row[1],
|
||||||
"location": row[2],
|
"location": row[2],
|
||||||
"number": (
|
"number": (
|
||||||
row[3].decode("utf-8") if row[3] and row[3] is bytes else row[3]
|
row[3].decode("utf-8")
|
||||||
|
if isinstance(row[3], bytes)
|
||||||
|
else row[3]
|
||||||
),
|
),
|
||||||
"provider": row[4],
|
"provider": row[4],
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -101,6 +101,11 @@ class SMS(IOSExtraction):
|
|||||||
ioc_match.message, "", result, matched_indicator=ioc_match.ioc
|
ioc_match.message, "", result, matched_indicator=ioc_match.ioc
|
||||||
)
|
)
|
||||||
|
|
||||||
|
def collect_url_results(self) -> None:
|
||||||
|
for message in self.results:
|
||||||
|
for url in message.get("links", []):
|
||||||
|
self.add_url_result(url, message.get("isodate"), "sms")
|
||||||
|
|
||||||
def run(self) -> None:
|
def run(self) -> None:
|
||||||
self._find_ios_database(backup_ids=SMS_BACKUP_IDS, root_paths=SMS_ROOT_PATHS)
|
self._find_ios_database(backup_ids=SMS_BACKUP_IDS, root_paths=SMS_ROOT_PATHS)
|
||||||
self.log.info("Found SMS database at path: %s", self.file_path)
|
self.log.info("Found SMS database at path: %s", self.file_path)
|
||||||
|
|||||||
@@ -70,6 +70,11 @@ class Whatsapp(IOSExtraction):
|
|||||||
ioc_match.message, "", result, matched_indicator=ioc_match.ioc
|
ioc_match.message, "", result, matched_indicator=ioc_match.ioc
|
||||||
)
|
)
|
||||||
|
|
||||||
|
def collect_url_results(self) -> None:
|
||||||
|
for message in self.results:
|
||||||
|
for url in message.get("links", []):
|
||||||
|
self.add_url_result(url, message.get("isodate"), "whatsapp")
|
||||||
|
|
||||||
def run(self) -> None:
|
def run(self) -> None:
|
||||||
self._find_ios_database(
|
self._find_ios_database(
|
||||||
backup_ids=WHATSAPP_BACKUP_IDS, root_paths=WHATSAPP_ROOT_PATHS
|
backup_ids=WHATSAPP_BACKUP_IDS, root_paths=WHATSAPP_ROOT_PATHS
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ import datetime
|
|||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from mvt.android.artifacts.tombstone_crashes import TombstoneCrashArtifact
|
from mvt.android.artifacts.tombstone_crashes import TombstoneCrashArtifact
|
||||||
|
from mvt.android.parsers.proto.tombstone import Tombstone
|
||||||
|
|
||||||
from ..utils import get_artifact
|
from ..utils import get_artifact
|
||||||
|
|
||||||
@@ -42,6 +43,47 @@ class TestTombstoneCrashArtifact:
|
|||||||
assert len(tombstone_artifact.results) == 1
|
assert len(tombstone_artifact.results) == 1
|
||||||
self.validate_tombstone_result(tombstone_artifact.results[0])
|
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):
|
def test_text_tombstone_keeps_crashing_thread(self):
|
||||||
tombstone_artifact = TombstoneCrashArtifact()
|
tombstone_artifact = TombstoneCrashArtifact()
|
||||||
artifact_path = "android_data/tombstone_process.txt"
|
artifact_path = "android_data/tombstone_process.txt"
|
||||||
|
|||||||
@@ -0,0 +1,371 @@
|
|||||||
|
from types import SimpleNamespace
|
||||||
|
|
||||||
|
import click
|
||||||
|
from click.testing import CliRunner
|
||||||
|
|
||||||
|
from mvt.common.cli_plugins import (
|
||||||
|
ANDROID_CLI_PLUGIN_GROUP,
|
||||||
|
IOS_CLI_PLUGIN_GROUP,
|
||||||
|
BrokenPluginCommand,
|
||||||
|
load_cli_commands_option,
|
||||||
|
register_cli_commands_from_path,
|
||||||
|
register_cli_plugins,
|
||||||
|
register_installed_cli_commands,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
COMMAND_TEMPLATE = """
|
||||||
|
import click
|
||||||
|
|
||||||
|
|
||||||
|
@click.command({name!r})
|
||||||
|
@click.pass_context
|
||||||
|
def cli(ctx):
|
||||||
|
click.echo({message!r})
|
||||||
|
if ctx.obj:
|
||||||
|
click.echo(ctx.obj.get("marker", ""))
|
||||||
|
"""
|
||||||
|
|
||||||
|
|
||||||
|
def _write_command(path, name, message="command ran"):
|
||||||
|
path.write_text(
|
||||||
|
COMMAND_TEMPLATE.format(name=name, message=message),
|
||||||
|
encoding="utf-8",
|
||||||
|
)
|
||||||
|
return path
|
||||||
|
|
||||||
|
|
||||||
|
def _make_group():
|
||||||
|
@click.group()
|
||||||
|
@load_cli_commands_option
|
||||||
|
@click.pass_context
|
||||||
|
def group(ctx):
|
||||||
|
ctx.ensure_object(dict)
|
||||||
|
ctx.obj["marker"] = "parent context"
|
||||||
|
|
||||||
|
return group
|
||||||
|
|
||||||
|
|
||||||
|
def _entry_point(name, value, command=None, exception=None, distribution="plugin"):
|
||||||
|
def load():
|
||||||
|
if exception is not None:
|
||||||
|
raise exception
|
||||||
|
return command
|
||||||
|
|
||||||
|
dist = SimpleNamespace(metadata={"Name": distribution}, version="1.0")
|
||||||
|
return SimpleNamespace(name=name, value=value, load=load, dist=dist)
|
||||||
|
|
||||||
|
|
||||||
|
def test_load_command_option_registers_command_before_resolution(tmp_path):
|
||||||
|
command_path = _write_command(tmp_path / "hello.py", "hello")
|
||||||
|
group = _make_group()
|
||||||
|
|
||||||
|
result = CliRunner().invoke(
|
||||||
|
group,
|
||||||
|
["--load-command", str(command_path), "hello"],
|
||||||
|
)
|
||||||
|
|
||||||
|
assert result.exit_code == 0
|
||||||
|
assert "command ran" in result.output
|
||||||
|
assert "parent context" in result.output
|
||||||
|
|
||||||
|
|
||||||
|
def test_load_command_option_supports_folders_and_repeated_paths(tmp_path):
|
||||||
|
folder = tmp_path / "commands"
|
||||||
|
folder.mkdir()
|
||||||
|
_write_command(folder / "b.py", "second")
|
||||||
|
_write_command(folder / "a.py", "first")
|
||||||
|
_write_command(folder / ".hidden.py", "hidden")
|
||||||
|
_write_command(folder / "__init__.py", "init")
|
||||||
|
other = _write_command(tmp_path / "third.py", "third")
|
||||||
|
group = _make_group()
|
||||||
|
|
||||||
|
result = CliRunner().invoke(
|
||||||
|
group,
|
||||||
|
[
|
||||||
|
"--load-command",
|
||||||
|
str(folder),
|
||||||
|
"--load-command",
|
||||||
|
str(other),
|
||||||
|
"--load-command",
|
||||||
|
str(other),
|
||||||
|
"--help",
|
||||||
|
],
|
||||||
|
)
|
||||||
|
|
||||||
|
assert result.exit_code == 0
|
||||||
|
assert "first" in result.output
|
||||||
|
assert "second" in result.output
|
||||||
|
assert "third" in result.output
|
||||||
|
assert "hidden" not in result.output
|
||||||
|
assert "init" not in result.output
|
||||||
|
|
||||||
|
|
||||||
|
def test_loaded_command_participates_in_shell_completion(tmp_path):
|
||||||
|
command_path = _write_command(tmp_path / "hello.py", "hello")
|
||||||
|
group = _make_group()
|
||||||
|
words = f"group --load-command {command_path} he"
|
||||||
|
|
||||||
|
result = CliRunner().invoke(
|
||||||
|
group,
|
||||||
|
[],
|
||||||
|
env={
|
||||||
|
"_GROUP_COMPLETE": "bash_complete",
|
||||||
|
"COMP_WORDS": words,
|
||||||
|
"COMP_CWORD": "3",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
assert result.exit_code == 0
|
||||||
|
assert "plain,hello" in result.output
|
||||||
|
|
||||||
|
|
||||||
|
def test_explicit_command_import_and_contract_failures_are_usage_errors(tmp_path):
|
||||||
|
broken_path = tmp_path / "broken.py"
|
||||||
|
broken_path.write_text("raise RuntimeError('broken import')", encoding="utf-8")
|
||||||
|
missing_cli_path = tmp_path / "missing_cli.py"
|
||||||
|
missing_cli_path.write_text("value = 1", encoding="utf-8")
|
||||||
|
|
||||||
|
broken_result = CliRunner().invoke(
|
||||||
|
_make_group(),
|
||||||
|
["--load-command", str(broken_path), "broken"],
|
||||||
|
)
|
||||||
|
missing_cli_result = CliRunner().invoke(
|
||||||
|
_make_group(),
|
||||||
|
["--load-command", str(missing_cli_path), "missing-cli"],
|
||||||
|
)
|
||||||
|
|
||||||
|
assert broken_result.exit_code == 2
|
||||||
|
assert "broken import" in broken_result.output
|
||||||
|
assert missing_cli_result.exit_code == 2
|
||||||
|
assert "must export a Click command or group named 'cli'" in (
|
||||||
|
missing_cli_result.output
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_environment_command_failure_gets_broken_placeholder(tmp_path):
|
||||||
|
command_path = tmp_path / "broken_command.py"
|
||||||
|
command_path.write_text("raise RuntimeError('broken import')", encoding="utf-8")
|
||||||
|
group = click.Group()
|
||||||
|
|
||||||
|
registered = register_cli_commands_from_path(group, command_path)
|
||||||
|
|
||||||
|
assert registered == ["broken-command"]
|
||||||
|
assert isinstance(group.commands["broken-command"], BrokenPluginCommand)
|
||||||
|
result = CliRunner().invoke(group, ["broken-command"])
|
||||||
|
assert result.exit_code == 1
|
||||||
|
assert "broken import" in result.output
|
||||||
|
assert str(command_path) in result.output
|
||||||
|
|
||||||
|
|
||||||
|
def test_environment_command_system_exit_gets_broken_placeholder(tmp_path):
|
||||||
|
command_path = tmp_path / "exiting_command.py"
|
||||||
|
command_path.write_text("raise SystemExit(7)", encoding="utf-8")
|
||||||
|
group = click.Group()
|
||||||
|
|
||||||
|
registered = register_cli_commands_from_path(group, command_path)
|
||||||
|
|
||||||
|
assert registered == ["exiting-command"]
|
||||||
|
assert isinstance(group.commands["exiting-command"], BrokenPluginCommand)
|
||||||
|
result = CliRunner().invoke(group, ["exiting-command"])
|
||||||
|
assert result.exit_code == 1
|
||||||
|
assert "Unable to import custom command" in result.output
|
||||||
|
assert result.output.rstrip().endswith(": 7")
|
||||||
|
|
||||||
|
|
||||||
|
def test_installed_entry_point_name_is_the_command_name(monkeypatch):
|
||||||
|
@click.command("internal-name")
|
||||||
|
def command():
|
||||||
|
click.echo("installed command ran")
|
||||||
|
|
||||||
|
entry_point = _entry_point(
|
||||||
|
"external-name",
|
||||||
|
"example_plugin:cli",
|
||||||
|
command=command,
|
||||||
|
)
|
||||||
|
monkeypatch.setattr(
|
||||||
|
"mvt.common.cli_plugins.importlib.metadata.entry_points",
|
||||||
|
lambda **kwargs: [entry_point],
|
||||||
|
)
|
||||||
|
group = click.Group()
|
||||||
|
|
||||||
|
registered = register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
|
||||||
|
|
||||||
|
assert registered == ["external-name"]
|
||||||
|
assert "internal-name" not in group.commands
|
||||||
|
result = CliRunner().invoke(group, ["external-name"])
|
||||||
|
assert result.exit_code == 0
|
||||||
|
assert result.output == "installed command ran\n"
|
||||||
|
|
||||||
|
|
||||||
|
def test_broken_installed_plugin_does_not_break_cli(monkeypatch):
|
||||||
|
broken = _entry_point(
|
||||||
|
"broken",
|
||||||
|
"broken_plugin:cli",
|
||||||
|
exception=RuntimeError("missing dependency"),
|
||||||
|
distribution="broken-plugin",
|
||||||
|
)
|
||||||
|
monkeypatch.setattr(
|
||||||
|
"mvt.common.cli_plugins.importlib.metadata.entry_points",
|
||||||
|
lambda **kwargs: [broken],
|
||||||
|
)
|
||||||
|
group = click.Group()
|
||||||
|
|
||||||
|
register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
|
||||||
|
|
||||||
|
help_result = CliRunner().invoke(group, ["--help"])
|
||||||
|
assert help_result.exit_code == 0
|
||||||
|
assert "Warning: external command could not be loaded." in help_result.output
|
||||||
|
|
||||||
|
result = CliRunner().invoke(group, ["broken"])
|
||||||
|
assert result.exit_code == 1
|
||||||
|
assert "broken-plugin 1.0 (broken_plugin:cli)" in result.output
|
||||||
|
assert "RuntimeError: missing dependency" in result.output
|
||||||
|
|
||||||
|
|
||||||
|
def test_installed_plugin_system_exit_does_not_break_cli(monkeypatch):
|
||||||
|
exiting = _entry_point(
|
||||||
|
"exiting",
|
||||||
|
"exiting_plugin:cli",
|
||||||
|
exception=SystemExit(7),
|
||||||
|
distribution="exiting-plugin",
|
||||||
|
)
|
||||||
|
monkeypatch.setattr(
|
||||||
|
"mvt.common.cli_plugins.importlib.metadata.entry_points",
|
||||||
|
lambda **kwargs: [exiting],
|
||||||
|
)
|
||||||
|
group = click.Group()
|
||||||
|
|
||||||
|
register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
|
||||||
|
|
||||||
|
help_result = CliRunner().invoke(group, ["--help"])
|
||||||
|
assert help_result.exit_code == 0
|
||||||
|
result = CliRunner().invoke(group, ["exiting"])
|
||||||
|
assert result.exit_code == 1
|
||||||
|
assert "SystemExit: 7" in result.output
|
||||||
|
|
||||||
|
|
||||||
|
def test_non_click_entry_point_gets_broken_placeholder(monkeypatch):
|
||||||
|
invalid = _entry_point("invalid", "plugin:value", command=object())
|
||||||
|
monkeypatch.setattr(
|
||||||
|
"mvt.common.cli_plugins.importlib.metadata.entry_points",
|
||||||
|
lambda **kwargs: [invalid],
|
||||||
|
)
|
||||||
|
group = click.Group()
|
||||||
|
|
||||||
|
register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
|
||||||
|
|
||||||
|
assert isinstance(group.commands["invalid"], BrokenPluginCommand)
|
||||||
|
result = CliRunner().invoke(group, ["invalid"])
|
||||||
|
assert result.exit_code == 1
|
||||||
|
assert "must resolve to a Click command or group" in result.output
|
||||||
|
|
||||||
|
|
||||||
|
def test_entry_point_discovery_failure_does_not_break_group(monkeypatch, caplog):
|
||||||
|
def fail_discovery(**kwargs):
|
||||||
|
raise RuntimeError("invalid package metadata")
|
||||||
|
|
||||||
|
monkeypatch.setattr(
|
||||||
|
"mvt.common.cli_plugins.importlib.metadata.entry_points",
|
||||||
|
fail_discovery,
|
||||||
|
)
|
||||||
|
group = click.Group()
|
||||||
|
|
||||||
|
registered = register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
|
||||||
|
|
||||||
|
assert registered == []
|
||||||
|
assert not group.commands
|
||||||
|
assert "Unable to discover external commands" in caplog.text
|
||||||
|
assert "invalid package metadata" in caplog.text
|
||||||
|
|
||||||
|
|
||||||
|
def test_builtin_and_first_external_command_win_collisions(monkeypatch, caplog):
|
||||||
|
@click.command("version")
|
||||||
|
def core_version():
|
||||||
|
pass
|
||||||
|
|
||||||
|
@click.command()
|
||||||
|
def first():
|
||||||
|
pass
|
||||||
|
|
||||||
|
@click.command()
|
||||||
|
def second():
|
||||||
|
pass
|
||||||
|
|
||||||
|
entry_points = [
|
||||||
|
_entry_point("duplicate", "z_plugin:cli", command=second, distribution="z"),
|
||||||
|
_entry_point("version", "plugin:version", command=first),
|
||||||
|
_entry_point("duplicate", "a_plugin:cli", command=first, distribution="a"),
|
||||||
|
]
|
||||||
|
monkeypatch.setattr(
|
||||||
|
"mvt.common.cli_plugins.importlib.metadata.entry_points",
|
||||||
|
lambda **kwargs: entry_points,
|
||||||
|
)
|
||||||
|
group = click.Group(commands={"version": core_version})
|
||||||
|
|
||||||
|
registered = register_installed_cli_commands(group, IOS_CLI_PLUGIN_GROUP)
|
||||||
|
|
||||||
|
assert registered == ["duplicate"]
|
||||||
|
assert group.commands["version"] is core_version
|
||||||
|
assert group.commands["duplicate"] is first
|
||||||
|
assert "the command name is already registered" in caplog.text
|
||||||
|
|
||||||
|
|
||||||
|
def test_explicit_command_cannot_replace_existing_command(tmp_path):
|
||||||
|
command_path = _write_command(tmp_path / "version.py", "version")
|
||||||
|
group = _make_group()
|
||||||
|
|
||||||
|
@group.command("version")
|
||||||
|
def core_version():
|
||||||
|
pass
|
||||||
|
|
||||||
|
result = CliRunner().invoke(
|
||||||
|
group,
|
||||||
|
["--load-command", str(command_path), "version"],
|
||||||
|
)
|
||||||
|
|
||||||
|
assert result.exit_code == 2
|
||||||
|
assert "the command name is already registered" in result.output
|
||||||
|
|
||||||
|
|
||||||
|
def test_platform_entry_point_groups_and_environment_paths_are_separate(
|
||||||
|
tmp_path, monkeypatch
|
||||||
|
):
|
||||||
|
ios_path = _write_command(tmp_path / "ios.py", "ios-file")
|
||||||
|
android_path = _write_command(tmp_path / "android.py", "android-file")
|
||||||
|
|
||||||
|
@click.command()
|
||||||
|
def ios_package():
|
||||||
|
pass
|
||||||
|
|
||||||
|
@click.command()
|
||||||
|
def android_package():
|
||||||
|
pass
|
||||||
|
|
||||||
|
def entry_points(*, group):
|
||||||
|
if group == IOS_CLI_PLUGIN_GROUP:
|
||||||
|
return [_entry_point("ios-package", "ios_plugin:cli", ios_package)]
|
||||||
|
return [_entry_point("android-package", "android_plugin:cli", android_package)]
|
||||||
|
|
||||||
|
monkeypatch.setattr(
|
||||||
|
"mvt.common.cli_plugins.importlib.metadata.entry_points",
|
||||||
|
entry_points,
|
||||||
|
)
|
||||||
|
monkeypatch.setenv("TEST_IOS_COMMANDS", str(ios_path))
|
||||||
|
monkeypatch.setenv("TEST_ANDROID_COMMANDS", str(android_path))
|
||||||
|
ios_group = click.Group()
|
||||||
|
android_group = click.Group()
|
||||||
|
|
||||||
|
register_cli_plugins(
|
||||||
|
ios_group,
|
||||||
|
entry_point_group=IOS_CLI_PLUGIN_GROUP,
|
||||||
|
environment_variable="TEST_IOS_COMMANDS",
|
||||||
|
)
|
||||||
|
register_cli_plugins(
|
||||||
|
android_group,
|
||||||
|
entry_point_group=ANDROID_CLI_PLUGIN_GROUP,
|
||||||
|
environment_variable="TEST_ANDROID_COMMANDS",
|
||||||
|
)
|
||||||
|
|
||||||
|
assert set(ios_group.commands) == {"ios-file", "ios-package"}
|
||||||
|
assert set(android_group.commands) == {"android-file", "android-package"}
|
||||||
@@ -42,6 +42,15 @@ class IndependentModule(RecordingModule):
|
|||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class URLRecordingModule(RecordingModule):
|
||||||
|
def collect_url_results(self):
|
||||||
|
self.add_url_result(
|
||||||
|
"https://example.org/message",
|
||||||
|
"2026-07-29 12:00:00.000000",
|
||||||
|
"test-chat",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
class CustomIOSBackupModule(RecordingModule):
|
class CustomIOSBackupModule(RecordingModule):
|
||||||
supported_commands = (("ios", "check-backup"),)
|
supported_commands = (("ios", "check-backup"),)
|
||||||
|
|
||||||
@@ -87,6 +96,21 @@ class TestCommand:
|
|||||||
alerts = json.loads((tmp_path / "alerts.json").read_text())
|
alerts = json.loads((tmp_path / "alerts.json").read_text())
|
||||||
assert alerts[0]["event"]["payload"] == "\\xa8\\xa9"
|
assert alerts[0]["event"]["payload"] == "\\xa8\\xa9"
|
||||||
|
|
||||||
|
def test_stores_collected_urls(self, tmp_path):
|
||||||
|
cmd = RecordingCommand(results_path=str(tmp_path))
|
||||||
|
cmd.modules = [URLRecordingModule]
|
||||||
|
|
||||||
|
cmd.run()
|
||||||
|
|
||||||
|
assert json.loads((tmp_path / "urls.json").read_text()) == [
|
||||||
|
{
|
||||||
|
"url": "https://example.org/message",
|
||||||
|
"expanded_url": None,
|
||||||
|
"timestamp": "2026-07-29 12:00:00.000000",
|
||||||
|
"source": "test-chat",
|
||||||
|
}
|
||||||
|
]
|
||||||
|
|
||||||
def test_modules_run_in_stable_topological_order(self):
|
def test_modules_run_in_stable_topological_order(self):
|
||||||
cmd = RecordingCommand()
|
cmd = RecordingCommand()
|
||||||
cmd.modules = [ThirdModule, IndependentModule, SecondModule, FirstModule]
|
cmd.modules = [ThirdModule, IndependentModule, SecondModule, FirstModule]
|
||||||
|
|||||||
@@ -197,6 +197,15 @@ class TestIndicators:
|
|||||||
assert matches[0] is None
|
assert matches[0] is None
|
||||||
assert matches[1]
|
assert matches[1]
|
||||||
assert matches[1].ioc.value == "example.org"
|
assert matches[1].ioc.value == "example.org"
|
||||||
|
assert (
|
||||||
|
ind.get_expanded_url("https://tinyurl.com/nested")
|
||||||
|
== "https://www.example.org/landing"
|
||||||
|
)
|
||||||
|
assert (
|
||||||
|
ind.get_expanded_url("https://t.co/nested")
|
||||||
|
== "https://www.example.org/landing"
|
||||||
|
)
|
||||||
|
assert ind.get_expanded_url("https://bit.ly/failure") is None
|
||||||
assert {call.args[0] for call in head.call_args_list} == {
|
assert {call.args[0] for call in head.call_args_list} == {
|
||||||
"https://bit.ly/failure",
|
"https://bit.ly/failure",
|
||||||
"https://tinyurl.com/nested",
|
"https://tinyurl.com/nested",
|
||||||
|
|||||||
@@ -18,6 +18,14 @@ class TestSMSModule:
|
|||||||
run_module(m)
|
run_module(m)
|
||||||
assert len(m.results) == 1
|
assert len(m.results) == 1
|
||||||
assert len(m.timeline) == 2
|
assert len(m.timeline) == 2
|
||||||
|
assert m.url_results == [
|
||||||
|
{
|
||||||
|
"url": "https://badbadbad.example.org/",
|
||||||
|
"expanded_url": None,
|
||||||
|
"timestamp": "2019-08-29 23:13:30.000000",
|
||||||
|
"source": "sms",
|
||||||
|
}
|
||||||
|
]
|
||||||
assert len(m.alertstore.alerts) == 0
|
assert len(m.alertstore.alerts) == 0
|
||||||
|
|
||||||
def test_detection(self, indicator_file):
|
def test_detection(self, indicator_file):
|
||||||
|
|||||||
@@ -0,0 +1,35 @@
|
|||||||
|
# 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 logging
|
||||||
|
|
||||||
|
from mvt.common.indicators import Indicators
|
||||||
|
from mvt.ios.modules.mixed.whatsapp import Whatsapp
|
||||||
|
|
||||||
|
|
||||||
|
def test_collect_url_results_includes_expansion():
|
||||||
|
module = Whatsapp(
|
||||||
|
results=[
|
||||||
|
{
|
||||||
|
"links": ["https://bit.ly/message"],
|
||||||
|
"isodate": "2026-07-29 12:00:00.000000",
|
||||||
|
}
|
||||||
|
]
|
||||||
|
)
|
||||||
|
module.indicators = Indicators(log=logging.getLogger())
|
||||||
|
module.indicators.resolved_urls["https://bit.ly/message"] = (
|
||||||
|
"https://example.org/landing"
|
||||||
|
)
|
||||||
|
|
||||||
|
module.collect_url_results()
|
||||||
|
|
||||||
|
assert module.url_results == [
|
||||||
|
{
|
||||||
|
"url": "https://bit.ly/message",
|
||||||
|
"expanded_url": "https://example.org/landing",
|
||||||
|
"timestamp": "2026-07-29 12:00:00.000000",
|
||||||
|
"source": "whatsapp",
|
||||||
|
}
|
||||||
|
]
|
||||||
@@ -33,6 +33,24 @@ class TestCheckAndroidqfCommand:
|
|||||||
result = runner.invoke(check_androidqf, [path])
|
result = runner.invoke(check_androidqf, [path])
|
||||||
assert result.exit_code == 0
|
assert result.exit_code == 0
|
||||||
|
|
||||||
|
def test_check_stores_nested_sms_urls(self, tmp_path):
|
||||||
|
runner = CliRunner()
|
||||||
|
path = os.path.join(get_artifact_folder(), "androidqf")
|
||||||
|
|
||||||
|
result = runner.invoke(check_androidqf, ["--output", str(tmp_path), path])
|
||||||
|
|
||||||
|
assert result.exit_code == 0
|
||||||
|
urls = json.loads((tmp_path / "urls.json").read_text())
|
||||||
|
assert {entry["url"] for entry in urls} == {
|
||||||
|
"http://google.com",
|
||||||
|
"https://google.com/",
|
||||||
|
}
|
||||||
|
assert all(
|
||||||
|
set(entry) == {"url", "expanded_url", "timestamp", "source"}
|
||||||
|
for entry in urls
|
||||||
|
)
|
||||||
|
assert all(entry["source"] == "sms" for entry in urls)
|
||||||
|
|
||||||
def test_acquisition_context_is_passed_to_bugreport(self, tmp_path, mocker):
|
def test_acquisition_context_is_passed_to_bugreport(self, tmp_path, mocker):
|
||||||
data_path = tmp_path / "androidqf"
|
data_path = tmp_path / "androidqf"
|
||||||
data_path.mkdir()
|
data_path.mkdir()
|
||||||
|
|||||||
@@ -1,5 +1,7 @@
|
|||||||
|
import io
|
||||||
import tarfile
|
import tarfile
|
||||||
from datetime import timedelta
|
from datetime import timedelta
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
from mvt.ios.cmd_check_sysdiagnose import CmdIOSCheckSysdiagnose
|
from mvt.ios.cmd_check_sysdiagnose import CmdIOSCheckSysdiagnose
|
||||||
from mvt.ios.modules.sysdiagnose import SysdiagnoseExtraction
|
from mvt.ios.modules.sysdiagnose import SysdiagnoseExtraction
|
||||||
@@ -70,6 +72,46 @@ def test_check_sysdiagnose_from_archive_closes_archive(tmp_path):
|
|||||||
{"content": "artifact", "timezone_offset": timedelta(hours=2).seconds}
|
{"content": "artifact", "timezone_offset": timedelta(hours=2).seconds}
|
||||||
]
|
]
|
||||||
assert command.executed[0].ips_files == [
|
assert command.executed[0].ips_files == [
|
||||||
{"file_path": "sysdiagnose/report.ips", "bug_type": 210}
|
{
|
||||||
|
"file_path": str(
|
||||||
|
Path(command.extracted_sysdiagnose_path) / "report.ips"
|
||||||
|
),
|
||||||
|
"bug_type": 210,
|
||||||
|
}
|
||||||
]
|
]
|
||||||
assert command.sysdiagnose_archive is None
|
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()
|
||||||
Reference in new issue
Block a user