mirror of
https://github.com/mvt-project/mvt.git
synced 2026-10-22 23:28:23 +02:00
Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3a90c0934a |
No files matched your search
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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: |
|
||||
@@ -59,24 +59,21 @@ jobs:
|
||||
type=raw,enable=${{ github.ref_type == 'tag' }},value=${{ github.ref_name }},suffix=${{ matrix.platform.tag-suffix }}
|
||||
type=sha,suffix=${{ matrix.platform.tag-suffix }}
|
||||
type=sha,format=long,suffix=${{ matrix.platform.tag-suffix }}
|
||||
# Register emulators so Buildx can build the ARM64 images on the AMD64 runner.
|
||||
- name: Set up QEMU
|
||||
uses: docker/setup-qemu-action@v4
|
||||
# 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: .
|
||||
platforms: linux/amd64,linux/arm64
|
||||
push: true
|
||||
tags: ${{ steps.meta.outputs.tags }}
|
||||
labels: ${{ steps.meta.outputs.labels }}
|
||||
provenance: mode=max
|
||||
sbom: true
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -11,8 +11,8 @@ jobs:
|
||||
update-ios-version:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v7
|
||||
- uses: actions/setup-python@v7
|
||||
- uses: actions/checkout@v6
|
||||
- uses: actions/setup-python@v6
|
||||
- name: Install script dependencies
|
||||
run: python -m pip install "packaging==26.3"
|
||||
- name: Run script to fetch latest iOS releases from Apple RSS feed.
|
||||
|
||||
+2
-13
@@ -119,23 +119,12 @@ ARG DEBIAN_FRONTEND=noninteractive
|
||||
RUN apt-get update \
|
||||
&& apt-get install -y \
|
||||
adb \
|
||||
binutils \
|
||||
default-jre-headless \
|
||||
file \
|
||||
jq \
|
||||
less \
|
||||
libcurl4 \
|
||||
libimage-exiftool-perl \
|
||||
libssl3 \
|
||||
libusb-1.0-0 \
|
||||
moreutils \
|
||||
p7zip-full \
|
||||
python3 \
|
||||
ripgrep \
|
||||
sqlite3 \
|
||||
tree \
|
||||
unzip \
|
||||
xxd
|
||||
sqlite3
|
||||
COPY --from=build-libplist /build /
|
||||
COPY --from=build-libimobiledevice-glue /build /
|
||||
COPY --from=build-libtatsu /build /
|
||||
@@ -155,7 +144,7 @@ ADD --checksum=sha256:a20e07f8b2ea47620aff0267f230c3f1f495f097081fd709eec51cf2a2
|
||||
RUN echo 'alias abe="java -jar /opt/abe/abe.jar"' >> ~/.bashrc
|
||||
|
||||
# Generate adb key folder
|
||||
RUN echo 'if [ ! -f /root/.android/adbkey ]; then adb keygen /root/.android/adbkey > /dev/null 2>&1; fi' >> ~/.bashrc
|
||||
RUN echo 'if [ ! -f /root/.android/adbkey ]; then adb keygen /root/.android/adbkey 2&>1 > /dev/null; fi' >> ~/.bashrc
|
||||
RUN mkdir /root/.android
|
||||
|
||||
# Setup investigations environment
|
||||
|
||||
+1
-1
@@ -31,7 +31,7 @@ ADD --checksum=sha256:a20e07f8b2ea47620aff0267f230c3f1f495f097081fd709eec51cf2a2
|
||||
RUN echo 'alias abe="java -jar /opt/abe/abe.jar"' >> ~/.bashrc
|
||||
|
||||
# Generate adb key folder
|
||||
RUN echo 'if [ ! -f /root/.android/adbkey ]; then adb keygen /root/.android/adbkey > /dev/null 2>&1; fi' >> ~/.bashrc
|
||||
RUN echo 'if [ ! -f /root/.android/adbkey ]; then adb keygen /root/.android/adbkey 2&>1 > /dev/null; fi' >> ~/.bashrc
|
||||
RUN mkdir /root/.android
|
||||
|
||||
ENTRYPOINT [ "/usr/local/bin/mvt-android" ]
|
||||
@@ -84,7 +84,7 @@ 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/development/custom_commands/)
|
||||
[custom CLI command documentation](https://docs.mvt.re/en/latest/custom_commands/)
|
||||
for the plugin entry points and `--load-command` interface.
|
||||
|
||||
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
# Custom CLI Commands
|
||||
|
||||
MVT can load additional top-level commands into `mvt-ios` and `mvt-android`.
|
||||
Custom commands are different from [custom forensic modules](index.md#custom-modules):
|
||||
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.
|
||||
|
||||
@@ -56,10 +56,6 @@ pipx inject mvt my-mvt-plugin
|
||||
When MVT is installed in an active virtual environment, install the plugin with
|
||||
`pip` in that environment.
|
||||
|
||||
Command packages that need their own settings, such as an API key, should store
|
||||
them in a namespaced [plugin configuration file](plugin_configuration.md)
|
||||
rather than in MVT's own `config.yaml`.
|
||||
|
||||
## Load a Command File
|
||||
|
||||
For local commands that are not packaged, create a Python file that exports one
|
||||
@@ -117,97 +117,6 @@ class DependentCustomModule(MVTModule):
|
||||
self.results = [{"manifest_entries": len(manifest_results)}]
|
||||
```
|
||||
|
||||
## Installed module packages
|
||||
|
||||
Python packages can register modules so they load automatically in every
|
||||
module-running `check-*` command, without `--load-module` or
|
||||
`MVT_CUSTOM_MODULES`. Register an entry point in the `mvt.modules` group in
|
||||
the package's `pyproject.toml`:
|
||||
|
||||
```toml
|
||||
[project.entry-points."mvt.modules"]
|
||||
mvt-plugin-amnesty-custom = "mvt_plugin_amnesty_custom:get_modules"
|
||||
```
|
||||
|
||||
The entry point must resolve to an iterable of `MVTModule` subclasses, or to
|
||||
a callable returning one:
|
||||
|
||||
```python
|
||||
from mvt.common.module import MVTModule
|
||||
|
||||
|
||||
class PackagedModule(MVTModule):
|
||||
supported_commands = (("ios", "check-backup"),)
|
||||
|
||||
def run(self):
|
||||
self.results = [{"message": "packaged module ran"}]
|
||||
|
||||
|
||||
def get_modules() -> list[type[MVTModule]]:
|
||||
return [PackagedModule]
|
||||
```
|
||||
|
||||
Installed modules follow the same rules as other custom modules: each module
|
||||
must declare `supported_commands`, and dependencies are resolved with the
|
||||
standard ordering logic. A broken entry point is skipped with a warning and
|
||||
does not prevent MVT from running. As with custom commands, installed module
|
||||
packages run as trusted code inside the MVT process, so install only packages
|
||||
from sources you trust.
|
||||
|
||||
For a `pipx` installation of MVT, inject the package into MVT's environment:
|
||||
|
||||
```bash
|
||||
pipx inject mvt mvt-plugin-amnesty-custom
|
||||
```
|
||||
|
||||
Module packages that need their own settings, such as an API key, should store
|
||||
them in a namespaced [plugin configuration file](plugin_configuration.md)
|
||||
rather than in MVT's own `config.yaml`.
|
||||
|
||||
### Naming module packages
|
||||
|
||||
Name module packages `mvt-plugin-<name>` (import package `mvt_plugin_<name>`),
|
||||
and include the name of the publishing organization or author so packages from
|
||||
different groups do not collide: for example, Amnesty International's custom
|
||||
modules would be distributed as `mvt-plugin-amnesty-custom` with the import
|
||||
package `mvt_plugin_amnesty_custom`.
|
||||
|
||||
The prefix makes module packages easy to find on PyPI and keeps their import
|
||||
names from clashing with unrelated Python packages. It is a convention, not a
|
||||
technical requirement: modules load through the `mvt.modules` entry point
|
||||
regardless of what the package is called, and existing packages with other
|
||||
names keep working. Note that the prefix is also not a mark of authenticity —
|
||||
anyone can publish a package with any available name, so vet a module package
|
||||
and its publisher before installing it, whatever it is called.
|
||||
|
||||
### Module logging
|
||||
|
||||
Modules log through `self.log`, and MVT names the logger for where the module
|
||||
came from. MVT's own modules log under their dotted path (for example
|
||||
`mvt.ios.modules.mixed.whatsapp`). Everything external is namespaced under
|
||||
`mvt.ext` to keep it visually distinct from built-in modules and isolated from
|
||||
MVT's internal logger tree:
|
||||
|
||||
- Installed packages log under `mvt.ext.<package>`, with the `mvt_plugin_`
|
||||
prefix stripped: modules in `mvt_plugin_amnesty_custom` log as
|
||||
`mvt.ext.amnesty_custom.*`.
|
||||
- Files loaded with `--load-module` or `MVT_CUSTOM_MODULES` log as
|
||||
`mvt.ext.<file name>`.
|
||||
|
||||
## Auditing loaded modules
|
||||
|
||||
Because installed module packages load automatically, MVT records where every
|
||||
module came from:
|
||||
|
||||
- `--list-modules` groups the available modules by source: MVT itself
|
||||
(with its version), each installed package (with its version and, when
|
||||
installed directly from a repository, the commit), and each file loaded
|
||||
with `--load-module` or `MVT_CUSTOM_MODULES` (with the SHA-256 hash of the
|
||||
file).
|
||||
- When a command runs with an `--output` folder, the `command.log` file
|
||||
records one line per module source with the source's version or hash and
|
||||
the list of modules loaded from it.
|
||||
|
||||
## Profiling
|
||||
|
||||
Some MVT modules extract and process significant amounts of data during the analysis process or while checking results against known indicators. Care must be
|
||||
@@ -1,149 +0,0 @@
|
||||
# Plugin Configuration
|
||||
|
||||
Plugin packages that add [custom CLI commands](custom_commands.md) or
|
||||
[modules](index.md#installed-module-packages) often need to store their
|
||||
own settings, such as an API key, a server URL or the timestamp of the last
|
||||
synchronization. MVT provides a namespaced settings base class so each plugin
|
||||
keeps its configuration in its own file, and a data folder for anything else a
|
||||
plugin needs to keep on disk.
|
||||
|
||||
!!! warning
|
||||
|
||||
Do not write plugin settings to MVT's own `config.yaml`. MVT rewrites that
|
||||
file with the settings it knows about every time it starts, so any other
|
||||
section is deleted.
|
||||
|
||||
## Where Settings Are Stored
|
||||
|
||||
Each plugin gets one YAML file in a `plugins` folder next to MVT's own
|
||||
configuration:
|
||||
|
||||
```
|
||||
~/.config/mvt/plugins/<plugin name>.yaml
|
||||
```
|
||||
|
||||
The exact parent folder follows the platform convention used for MVT's
|
||||
`config.yaml` (for example `~/Library/Application Support/mvt` on macOS). Use
|
||||
`mvt.common.plugin_config.plugin_config_path()` instead of building the path by
|
||||
hand.
|
||||
|
||||
Plugin names must be lowercase and may only contain letters, digits and dashes,
|
||||
matching the `mvt-plugin-<name>` package naming convention. MVT creates the
|
||||
`plugins` folder with `0700` permissions and writes the settings files with
|
||||
`0600` permissions, because they commonly hold credentials. Files are written
|
||||
through a temporary file and moved into place, so an interrupted save never
|
||||
leaves a partially written settings file behind.
|
||||
|
||||
## Plugin Data Folder
|
||||
|
||||
Everything else a plugin keeps on disk, such as a cache, a downloaded artifact
|
||||
or synchronization state, belongs in the folder returned by the `data_folder()`
|
||||
class method of the plugin's settings class, or by
|
||||
`mvt.common.plugin_config.plugin_data_folder()` called with the plugin name if
|
||||
the plugin has no settings class:
|
||||
|
||||
```
|
||||
~/.local/share/mvt/plugin-data/<plugin name>/ # Linux
|
||||
~/Library/Application Support/mvt/plugin-data/<plugin name>/ # macOS
|
||||
```
|
||||
|
||||
The folder sits beside MVT's own data, such as the downloaded indicators. It is
|
||||
created if it is missing, with `0700` permissions. Asking for it again returns
|
||||
the same path and leaves the contents alone, so a plugin can ask for it every
|
||||
time it needs the folder. `ExamplePluginSettings` below is the settings class
|
||||
defined in the next section:
|
||||
|
||||
```python
|
||||
import os
|
||||
|
||||
|
||||
def cache_path() -> str:
|
||||
folder = ExamplePluginSettings.data_folder()
|
||||
return os.path.join(folder, "virustotal_lookups_cache.json")
|
||||
```
|
||||
|
||||
A plugin which has no settings class calls
|
||||
`plugin_data_folder("example-plugin")` instead.
|
||||
|
||||
Do not fall back on a path of your own such as `~/.cache/example-plugin`: it
|
||||
is a Linux-only convention, and MVT will not create it for you.
|
||||
|
||||
## Defining Plugin Settings
|
||||
|
||||
Subclass `MVTPluginSettings`, set `plugin_name` and declare typed fields with
|
||||
defaults:
|
||||
|
||||
```python
|
||||
from typing import Optional
|
||||
|
||||
from mvt.common.plugin_config import MVTPluginSettings
|
||||
|
||||
|
||||
class ExamplePluginSettings(MVTPluginSettings):
|
||||
plugin_name = "example-plugin"
|
||||
|
||||
API_KEY: Optional[str] = None
|
||||
MAX_RESULTS: int = 25
|
||||
LAST_SYNC: Optional[str] = None
|
||||
```
|
||||
|
||||
`load()` returns the current settings and `save()` writes them back:
|
||||
|
||||
```python
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import click
|
||||
|
||||
|
||||
def sync():
|
||||
settings = ExamplePluginSettings.load()
|
||||
if not settings.API_KEY:
|
||||
raise click.ClickException(
|
||||
"No API key configured. Set MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY or "
|
||||
"run 'example-plugin configure'."
|
||||
)
|
||||
|
||||
settings.LAST_SYNC = datetime.now(timezone.utc).isoformat()
|
||||
settings.save()
|
||||
```
|
||||
|
||||
A missing settings file is not an error: the plugin then runs on the field
|
||||
defaults and on whatever the environment provides. `save()` only persists the
|
||||
values that differ from the defaults, and it never touches MVT's `config.yaml`.
|
||||
A settings file that cannot be parsed, or that does not hold a mapping of
|
||||
setting names to values, raises a `PluginConfigLoadError` naming the file.
|
||||
|
||||
Every subclass that sets its own `plugin_name` gets its own file and its own
|
||||
environment namespace. A subclass that does not redefine `plugin_name` inherits
|
||||
it, and therefore shares the file and the environment variables of its parent
|
||||
class.
|
||||
|
||||
## Environment Variables
|
||||
|
||||
Every field can also be set with an environment variable. The prefix is
|
||||
`MVT_PLUGIN_`, followed by the plugin name upper-cased with dashes replaced by
|
||||
underscores, followed by the field name. For the example above:
|
||||
|
||||
```bash
|
||||
export MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY=...
|
||||
export MVT_PLUGIN_EXAMPLE_PLUGIN_MAX_RESULTS=50
|
||||
```
|
||||
|
||||
Settings resolve in this order, from highest to lowest priority:
|
||||
|
||||
1. Arguments passed to the settings class directly, such as
|
||||
`ExamplePluginSettings(API_KEY="...")`
|
||||
2. Environment variables
|
||||
3. The plugin's YAML file
|
||||
4. The field defaults declared on the settings class
|
||||
|
||||
!!! tip
|
||||
|
||||
On shared or multi-user machines, prefer passing API keys through
|
||||
environment variables rather than saving them to the plugin file. `save()`
|
||||
skips every value that the environment currently supplies, so a credential
|
||||
provided that way is not copied into the settings file when a plugin saves
|
||||
an unrelated setting.
|
||||
|
||||
Unknown keys in a plugin's YAML file are ignored, so a settings file written by
|
||||
a newer version of a plugin does not break an older one.
|
||||
+1
-15
@@ -31,18 +31,4 @@ Test if the image was created successfully:
|
||||
docker run -it mvt
|
||||
```
|
||||
|
||||
If a prompt is spawned successfully, you can close it with `exit`.
|
||||
|
||||
## Access an iOS device from Docker
|
||||
|
||||
On the Linux host, install and start [usbmuxd](https://github.com/libimobiledevice/usbmuxd), then connect and unlock the iOS device. The daemon exposes the device through the `/var/run/usbmuxd` socket.
|
||||
|
||||
Bind that socket into the container to let MVT communicate with the device:
|
||||
|
||||
```bash
|
||||
docker run -it \
|
||||
--mount type=bind,source=/var/run/usbmuxd,target=/var/run/usbmuxd \
|
||||
ghcr.io/mvt-project/mvt
|
||||
```
|
||||
|
||||
If you built the image from source, replace `ghcr.io/mvt-project/mvt` with `mvt`.
|
||||
If a prompt is spawned successfully, you can close it with `exit`.
|
||||
+1
-15
@@ -417,21 +417,7 @@ If indicators are provided through the command-line, they are checked against th
|
||||
Backup: :material-check:
|
||||
Full filesystem dump: :material-check:
|
||||
|
||||
This JSON file is created by mvt-ios' `WhatsApp` module. The module extracts a list of WhatsApp messages from the SQLite database located at *private/var/mobile/Containers/Shared/AppGroup/\*/ChatStorage.sqlite*, along with one record per chat session (marked with `"record_type": "chat_session"`) containing the first and last interaction dates of each conversation. Chat sessions produce `chat_first_message` and `chat_last_message` timeline events, and group chats additionally produce a `group_created` event. A chat session's last-message date can postdate its newest stored message when the most recent messages in the chat were deleted.
|
||||
|
||||
Recent WhatsApp versions key 1:1 chat sessions by an opaque LID identifier rather than the contact's phone number. The module resolves these using the `ZWAPHONENUMBERLIDPAIR` table from the *LID.sqlite* database in the same app group (or from *ChatStorage.sqlite* itself in versions that store it there), populating `partner_resolved_phone_number` on chat session records and using the phone number in timeline events. Each LID-phone number pair is also extracted as a record (`"record_type": "lid_phone_number_pair"`) and produces a `lid_pair_recorded` timeline event marking when WhatsApp learned the association.
|
||||
This JSON file is created by mvt-ios' `WhatsApp` module. The module extracts a list of WhatsApp messages from the SQLite database located at *private/var/mobile/Containers/Shared/AppGroup/\*/ChatStorage.sqlite*.
|
||||
|
||||
If indicators are provided through the command-line, they are checked against the extracted HTTP links. Any matches are stored in *whatsapp_detected.json*.
|
||||
|
||||
---
|
||||
|
||||
### `whatsapp_contacts.json`
|
||||
|
||||
!!! info "Availability"
|
||||
Backup: :material-check:
|
||||
Full filesystem dump: :material-check:
|
||||
|
||||
This JSON file is created by mvt-ios' `WhatsappContacts` module. The module extracts WhatsApp contact records from the SQLite database located at *private/var/mobile/Containers/Shared/AppGroup/\*/ContactsV2.sqlite*, including each contact's phone number, WhatsApp and LID identifiers, and the per-contact disappearing messages timer, which is not recorded in *ChatStorage.sqlite*. Each timestamp stored on a contact record produces a timeline event: `disappearing_mode_set` (when the disappearing messages timer was last changed), `about_changed` (when the contact last changed their "about" text), `about_expiration` (when a timed "about" is scheduled to expire) and `contact_last_updated` (when the contact record was last updated).
|
||||
|
||||
This database is often missing from incremental backups. When it cannot be found, the module logs a warning and produces no results, in which case the disappearing messages state of chats cannot be determined from the backup.
|
||||
|
||||
+2
-4
@@ -31,6 +31,7 @@ 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"
|
||||
@@ -49,8 +50,5 @@ nav:
|
||||
- Check an Android Backup (SMS messages): "android/backup.md"
|
||||
- Check Android Intrusion Logs: "android/intrusion_logs.md"
|
||||
- Indicators of Compromise: "iocs.md"
|
||||
- Development:
|
||||
- Development Instructions: "development/index.md"
|
||||
- Custom CLI Commands: "development/custom_commands.md"
|
||||
- Plugin Configuration: "development/plugin_configuration.md"
|
||||
- Development: "development.md"
|
||||
- License: "license.md"
|
||||
@@ -4,7 +4,7 @@
|
||||
# https://license.mvt.re/1.1/
|
||||
|
||||
import re
|
||||
from typing import Any, Dict, List, Optional
|
||||
from typing import Any, Dict, List
|
||||
|
||||
from mvt.android.utils import ROOT_PACKAGES
|
||||
from mvt.common.module_types import ModuleAtomicResult, ModuleSerializedResult
|
||||
@@ -71,8 +71,6 @@ class DumpsysPackagesArtifact(AndroidArtifact):
|
||||
"timestamp": "",
|
||||
"first_install_time": "",
|
||||
"last_update_time": "",
|
||||
"installer": "",
|
||||
"system": False,
|
||||
"permissions": list(),
|
||||
"requested_permissions": list(),
|
||||
}
|
||||
@@ -80,14 +78,7 @@ class DumpsysPackagesArtifact(AndroidArtifact):
|
||||
in_runtime_permissions = False
|
||||
in_declared_permissions = False
|
||||
in_requested_permissions = True
|
||||
current_user: Optional[int] = None
|
||||
first_install_times: Dict[Optional[int], str] = {}
|
||||
runtime_permissions: Dict[Optional[int], List[Dict[str, Any]]] = {}
|
||||
for line in output.splitlines():
|
||||
user_match = re.match(r"User (\d+):", line.strip())
|
||||
if user_match:
|
||||
current_user = int(user_match.group(1))
|
||||
|
||||
if in_install_permissions:
|
||||
if line.startswith(" " * 4) and not line.startswith(" " * 6):
|
||||
in_install_permissions = False
|
||||
@@ -111,7 +102,7 @@ class DumpsysPackagesArtifact(AndroidArtifact):
|
||||
if "granted=" in lineinfo[1]:
|
||||
granted = "granted=true" in lineinfo[1]
|
||||
|
||||
runtime_permissions.setdefault(current_user, []).append(
|
||||
details["permissions"].append(
|
||||
{"name": permission, "granted": granted, "type": "runtime"}
|
||||
)
|
||||
if in_declared_permissions:
|
||||
@@ -135,12 +126,8 @@ class DumpsysPackagesArtifact(AndroidArtifact):
|
||||
details["version_code"] = line.split("=", 1)[1].strip()
|
||||
elif line.strip().startswith("timeStamp="):
|
||||
details["timestamp"] = line.split("=")[1].strip()
|
||||
elif line.strip().startswith("installerPackageName="):
|
||||
details["installer"] = line.split("=", 1)[1].strip()
|
||||
elif line.strip().startswith("pkgFlags="):
|
||||
details["system"] = "SYSTEM" in line.split("=", 1)[1].split()
|
||||
elif line.strip().startswith("firstInstallTime="):
|
||||
first_install_times[current_user] = line.split("=", 1)[1].strip()
|
||||
details["first_install_time"] = line.split("=")[1].strip()
|
||||
elif line.strip().startswith("lastUpdateTime="):
|
||||
details["last_update_time"] = line.split("=")[1].strip()
|
||||
elif line.strip() == "install permissions:":
|
||||
@@ -152,19 +139,6 @@ class DumpsysPackagesArtifact(AndroidArtifact):
|
||||
elif line.strip() == "requested permissions:":
|
||||
in_requested_permissions = True
|
||||
|
||||
if 0 in first_install_times:
|
||||
details["first_install_time"] = first_install_times[0]
|
||||
elif None in first_install_times:
|
||||
details["first_install_time"] = first_install_times[None]
|
||||
elif first_install_times:
|
||||
details["first_install_time"] = next(iter(first_install_times.values()))
|
||||
|
||||
if 0 in runtime_permissions:
|
||||
details["permissions"].extend(runtime_permissions[0])
|
||||
elif None in runtime_permissions:
|
||||
details["permissions"].extend(runtime_permissions[None])
|
||||
elif runtime_permissions:
|
||||
details["permissions"].extend(next(iter(runtime_permissions.values())))
|
||||
return details
|
||||
|
||||
def parse_dumpsys_packages(self, output: str) -> List[Dict[str, Any]]:
|
||||
|
||||
@@ -9,7 +9,6 @@ from typing import Optional
|
||||
|
||||
from mvt.common.command import Command
|
||||
from mvt.common.module import MVTModule
|
||||
from mvt.common.module_loader import get_module_logger
|
||||
from mvt.common.utils import exec_or_profile
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
@@ -77,7 +76,7 @@ class CmdCheckIOCS(Command):
|
||||
)
|
||||
|
||||
m = iocs_module.from_json(
|
||||
file_path, log=get_module_logger(iocs_module)
|
||||
file_path, log=logging.getLogger(iocs_module.__module__)
|
||||
)
|
||||
if not m:
|
||||
log.warning("No result from this module, skipping it")
|
||||
|
||||
@@ -19,12 +19,7 @@ from .alerts import AlertLevel, AlertStore
|
||||
from .config import settings
|
||||
from .indicators import Indicators
|
||||
from .module import EncryptedBackupError, MVTModule, run_module, save_timeline
|
||||
from .module_loader import (
|
||||
ModuleOrigin,
|
||||
get_module_logger,
|
||||
get_module_origin,
|
||||
module_supports_command,
|
||||
)
|
||||
from .module_loader import module_supports_command
|
||||
from .module_types import ModuleTimeline, URLResult
|
||||
from .utils import (
|
||||
CustomJSONEncoder,
|
||||
@@ -215,34 +210,10 @@ class Command:
|
||||
for file in generate_hashes_from_path(self.target_path, self.log):
|
||||
self.hash_values.append(file)
|
||||
|
||||
@staticmethod
|
||||
def _modules_by_origin(
|
||||
modules: list[type[MVTModule]],
|
||||
) -> dict[ModuleOrigin, list[str]]:
|
||||
origins: dict[ModuleOrigin, list[str]] = {}
|
||||
for module in modules:
|
||||
origins.setdefault(get_module_origin(module), []).append(module.__name__)
|
||||
return origins
|
||||
|
||||
def list_modules(self) -> None:
|
||||
self.log.info("Following is the list of available %s modules:", self.name)
|
||||
for origin, module_names in self._modules_by_origin(
|
||||
self._available_modules()
|
||||
).items():
|
||||
self.log.info(
|
||||
" - Modules from %s: %s", origin.label, ", ".join(module_names)
|
||||
)
|
||||
|
||||
def _log_loaded_modules(self, modules: list[type[MVTModule]]) -> None:
|
||||
"""Record the loaded modules and their origins for auditability."""
|
||||
for origin, module_names in self._modules_by_origin(modules).items():
|
||||
self.log.info(
|
||||
"Loaded %d %s modules from %s: %s",
|
||||
len(module_names),
|
||||
self.name,
|
||||
origin.label,
|
||||
", ".join(module_names),
|
||||
)
|
||||
for module in self._available_modules():
|
||||
self.log.info(" - %s", module.__name__)
|
||||
|
||||
def _available_modules(self) -> list[type[MVTModule]]:
|
||||
modules = list(self.modules)
|
||||
@@ -389,8 +360,6 @@ class Command:
|
||||
if ordered_modules is None:
|
||||
return
|
||||
|
||||
self._log_loaded_modules(ordered_modules)
|
||||
|
||||
try:
|
||||
self.init()
|
||||
except NotImplementedError:
|
||||
@@ -399,7 +368,7 @@ class Command:
|
||||
executed_by_type: dict[type[MVTModule], MVTModule] = {}
|
||||
for module in ordered_modules:
|
||||
|
||||
module_logger = get_module_logger(module)
|
||||
module_logger = logging.getLogger(module.__module__)
|
||||
|
||||
m = module(
|
||||
target_path=self.target_path,
|
||||
|
||||
@@ -8,7 +8,7 @@ HELP_MSG_VERSION = "Show the currently installed version of MVT"
|
||||
HELP_MSG_OUTPUT = "Specify a path to a folder where you want to store JSON results"
|
||||
HELP_MSG_IOC = "Path to indicators file (can be invoked multiple time)"
|
||||
HELP_MSG_FAST = "Avoid running time/resource consuming features"
|
||||
HELP_MSG_LIST_MODULES = "Print list of available modules and their source, then exit"
|
||||
HELP_MSG_LIST_MODULES = "Print list of available modules and exit"
|
||||
HELP_MSG_MODULE = "Name of a single module you would like to run instead of all"
|
||||
HELP_MSG_LOAD_MODULE = (
|
||||
"Load custom MVT module(s) from a Python file or folder "
|
||||
|
||||
@@ -4,28 +4,18 @@
|
||||
# https://license.mvt.re/1.1/
|
||||
|
||||
import hashlib
|
||||
import importlib.metadata
|
||||
import importlib.util
|
||||
import inspect
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
from functools import lru_cache
|
||||
from pathlib import Path
|
||||
from types import ModuleType
|
||||
from typing import Iterable, Optional
|
||||
|
||||
from .module import MVTModule
|
||||
from .version import MVT_VERSION
|
||||
|
||||
MVT_CUSTOM_MODULES_ENV = "MVT_CUSTOM_MODULES"
|
||||
MODULES_ENTRY_POINT_GROUP = "mvt.modules"
|
||||
EXTERNAL_LOGGER_NAMESPACE = "mvt.ext"
|
||||
PLUGIN_PACKAGE_PREFIX = "mvt_plugin_"
|
||||
_ORIGIN_ATTRIBUTE = "_mvt_module_origin"
|
||||
_PATH_MODULE_PREFIX = "_mvt_custom_module_"
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@@ -33,74 +23,15 @@ class CustomModuleLoadError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ModuleOrigin:
|
||||
"""Describes where a loaded module came from, for auditability.
|
||||
|
||||
``kind`` is one of ``builtin`` (shipped with MVT), ``package`` (loaded
|
||||
from an installed package) or ``path`` (loaded from a file passed with
|
||||
``--load-module`` or the environment variable).
|
||||
"""
|
||||
|
||||
kind: str
|
||||
name: str
|
||||
version: Optional[str] = None
|
||||
commit: Optional[str] = None
|
||||
file_sha256: Optional[str] = None
|
||||
|
||||
@property
|
||||
def label(self) -> str:
|
||||
label = self.name
|
||||
if self.version:
|
||||
label += f"@{self.version}"
|
||||
label = f"'{label}'"
|
||||
if self.commit:
|
||||
label += f" (commit {self.commit})"
|
||||
if self.file_sha256:
|
||||
label += f" (sha256: {self.file_sha256})"
|
||||
return label
|
||||
|
||||
|
||||
def _module_name_for_path(path: Path) -> str:
|
||||
digest = hashlib.sha256(str(path).encode("utf-8")).hexdigest()[:16]
|
||||
return f"{_PATH_MODULE_PREFIX}{path.stem}_{digest}"
|
||||
|
||||
|
||||
def get_module_logger(module_class: type[MVTModule]) -> logging.Logger:
|
||||
"""Return the logger a module's records should be emitted through.
|
||||
|
||||
Modules loaded from installed packages or file paths live outside the
|
||||
"mvt" logger hierarchy, so their records would never reach the handlers
|
||||
attached to the "mvt" logger and instead fall through to
|
||||
logging.lastResort (which prints bare messages and drops anything below
|
||||
WARNING). Their loggers are parented under the "mvt.ext" namespace,
|
||||
keeping external module names from colliding with MVT's own logger
|
||||
tree. File-path modules are named after their file instead of the
|
||||
mangled internal import name, and packages following the recommended
|
||||
"mvt_plugin_<name>" naming convention log under "mvt.ext.<name>".
|
||||
"""
|
||||
name = module_class.__module__
|
||||
if name == "mvt" or name.startswith("mvt."):
|
||||
return logging.getLogger(name)
|
||||
|
||||
if name.startswith(_PATH_MODULE_PREFIX):
|
||||
name = Path(get_module_origin(module_class).name).stem
|
||||
else:
|
||||
top_level, separator, rest = name.partition(".")
|
||||
if top_level.startswith(PLUGIN_PACKAGE_PREFIX) and len(top_level) > len(
|
||||
PLUGIN_PACKAGE_PREFIX
|
||||
):
|
||||
name = top_level[len(PLUGIN_PACKAGE_PREFIX) :] + separator + rest
|
||||
|
||||
return logging.getLogger(f"{EXTERNAL_LOGGER_NAMESPACE}.{name}")
|
||||
return f"_mvt_custom_module_{path.stem}_{digest}"
|
||||
|
||||
|
||||
def _iter_module_files(path: Path) -> Iterable[Path]:
|
||||
if path.is_file():
|
||||
if path.suffix != ".py":
|
||||
raise CustomModuleLoadError(
|
||||
f"Custom module file is not a Python file: {path}"
|
||||
)
|
||||
raise CustomModuleLoadError(f"Custom module file is not a Python file: {path}")
|
||||
yield path
|
||||
return
|
||||
|
||||
@@ -128,9 +59,7 @@ def _load_python_file(path: Path) -> ModuleType:
|
||||
try:
|
||||
spec.loader.exec_module(module)
|
||||
except Exception as exc:
|
||||
raise CustomModuleLoadError(
|
||||
f"Unable to import custom module {path}: {exc}"
|
||||
) from exc
|
||||
raise CustomModuleLoadError(f"Unable to import custom module {path}: {exc}") from exc
|
||||
|
||||
return module
|
||||
|
||||
@@ -155,156 +84,17 @@ def load_custom_modules_from_path(path: str) -> list[type[MVTModule]]:
|
||||
resolved_path = Path(path).expanduser().resolve()
|
||||
|
||||
for module_file in _iter_module_files(resolved_path):
|
||||
file_sha256 = hashlib.sha256(module_file.read_bytes()).hexdigest()
|
||||
loaded_module = _load_python_file(module_file)
|
||||
origin = ModuleOrigin(
|
||||
kind="path", name=str(module_file), file_sha256=file_sha256
|
||||
)
|
||||
for module_class in discover_mvt_modules(loaded_module):
|
||||
key = (str(module_file), module_class.__qualname__)
|
||||
if key in seen:
|
||||
continue
|
||||
seen.add(key)
|
||||
setattr(module_class, _ORIGIN_ATTRIBUTE, origin)
|
||||
custom_modules.append(module_class)
|
||||
|
||||
return custom_modules
|
||||
|
||||
|
||||
def _module_key(module_class: type[MVTModule]) -> tuple[str, str]:
|
||||
try:
|
||||
source = str(Path(inspect.getfile(module_class)).resolve())
|
||||
except (OSError, TypeError):
|
||||
source = module_class.__module__
|
||||
return (source, module_class.__qualname__)
|
||||
|
||||
|
||||
def _distribution_commit(dist: importlib.metadata.Distribution) -> Optional[str]:
|
||||
"""Return the VCS commit a distribution was installed from, if recorded.
|
||||
|
||||
Packages installed directly from a repository (``pip install git+...``)
|
||||
record the commit in ``direct_url.json`` (PEP 610).
|
||||
"""
|
||||
try:
|
||||
direct_url_text = dist.read_text("direct_url.json")
|
||||
if not direct_url_text:
|
||||
return None
|
||||
commit = json.loads(direct_url_text).get("vcs_info", {}).get("commit_id")
|
||||
return commit if isinstance(commit, str) else None
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def _entry_point_origin(entry_point: importlib.metadata.EntryPoint) -> ModuleOrigin:
|
||||
name = entry_point.name
|
||||
version = None
|
||||
commit = None
|
||||
# Manually constructed entry points have no associated distribution.
|
||||
dist = getattr(entry_point, "dist", None)
|
||||
if dist is not None:
|
||||
try:
|
||||
name = dist.name or name
|
||||
version = dist.version
|
||||
except Exception:
|
||||
pass
|
||||
commit = _distribution_commit(dist)
|
||||
return ModuleOrigin(kind="package", name=name, version=version, commit=commit)
|
||||
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def _packages_distributions() -> dict[str, list[str]]:
|
||||
try:
|
||||
return dict(importlib.metadata.packages_distributions())
|
||||
except Exception:
|
||||
return {}
|
||||
|
||||
|
||||
def get_module_origin(module_class: type[MVTModule]) -> ModuleOrigin:
|
||||
"""Return the origin of a module class for auditing purposes."""
|
||||
origin = module_class.__dict__.get(_ORIGIN_ATTRIBUTE)
|
||||
if isinstance(origin, ModuleOrigin):
|
||||
return origin
|
||||
|
||||
top_level = module_class.__module__.partition(".")[0]
|
||||
if top_level == "mvt":
|
||||
return ModuleOrigin(kind="builtin", name="mvt", version=MVT_VERSION)
|
||||
|
||||
distributions = _packages_distributions().get(top_level)
|
||||
if distributions:
|
||||
name = distributions[0]
|
||||
version = None
|
||||
commit = None
|
||||
try:
|
||||
dist = importlib.metadata.distribution(name)
|
||||
version = dist.version
|
||||
commit = _distribution_commit(dist)
|
||||
except Exception:
|
||||
pass
|
||||
return ModuleOrigin(kind="package", name=name, version=version, commit=commit)
|
||||
|
||||
try:
|
||||
source = str(Path(inspect.getfile(module_class)).resolve())
|
||||
except (OSError, TypeError):
|
||||
source = module_class.__module__
|
||||
return ModuleOrigin(kind="path", name=source)
|
||||
|
||||
|
||||
def load_installed_modules() -> list[type[MVTModule]]:
|
||||
"""Load MVT modules registered by installed packages.
|
||||
|
||||
Packages register modules in the ``mvt.modules`` entry-point group. Each
|
||||
entry point must resolve to an iterable of MVTModule subclasses, or to a
|
||||
callable which returns one. A broken entry point is skipped with a
|
||||
warning so that a faulty plugin package cannot break MVT.
|
||||
"""
|
||||
try:
|
||||
entry_points = importlib.metadata.entry_points(group=MODULES_ENTRY_POINT_GROUP)
|
||||
except Exception as exc:
|
||||
log.warning(
|
||||
"Unable to discover installed module packages in entry-point group %s: %s",
|
||||
MODULES_ENTRY_POINT_GROUP,
|
||||
exc,
|
||||
)
|
||||
return []
|
||||
|
||||
installed_modules: list[type[MVTModule]] = []
|
||||
ordered_entry_points = sorted(
|
||||
entry_points, key=lambda entry_point: (entry_point.name, entry_point.value)
|
||||
)
|
||||
for entry_point in ordered_entry_points:
|
||||
try:
|
||||
loaded = entry_point.load()
|
||||
if callable(loaded) and not isinstance(loaded, type):
|
||||
loaded = loaded()
|
||||
module_classes = list(loaded)
|
||||
except (Exception, SystemExit) as exc:
|
||||
log.warning(
|
||||
"Unable to load modules from entry point %s (%s): %s",
|
||||
entry_point.name,
|
||||
entry_point.value,
|
||||
exc,
|
||||
)
|
||||
continue
|
||||
|
||||
origin = _entry_point_origin(entry_point)
|
||||
for module_class in module_classes:
|
||||
if not (
|
||||
isinstance(module_class, type) and issubclass(module_class, MVTModule)
|
||||
):
|
||||
log.warning(
|
||||
"Entry point %s (%s) provided %r which is not an "
|
||||
"MVTModule subclass",
|
||||
entry_point.name,
|
||||
entry_point.value,
|
||||
module_class,
|
||||
)
|
||||
continue
|
||||
setattr(module_class, _ORIGIN_ATTRIBUTE, origin)
|
||||
installed_modules.append(module_class)
|
||||
|
||||
return installed_modules
|
||||
|
||||
|
||||
def load_custom_modules(paths: Optional[Iterable[str]] = None) -> list[type[MVTModule]]:
|
||||
search_paths: list[str] = []
|
||||
env_path = os.environ.get(MVT_CUSTOM_MODULES_ENV)
|
||||
@@ -315,17 +105,10 @@ def load_custom_modules(paths: Optional[Iterable[str]] = None) -> list[type[MVTM
|
||||
|
||||
custom_modules: list[type[MVTModule]] = []
|
||||
seen: set[tuple[str, str]] = set()
|
||||
|
||||
for module_class in load_installed_modules():
|
||||
key = _module_key(module_class)
|
||||
if key in seen:
|
||||
continue
|
||||
seen.add(key)
|
||||
custom_modules.append(module_class)
|
||||
|
||||
for path in search_paths:
|
||||
for module_class in load_custom_modules_from_path(path):
|
||||
key = _module_key(module_class)
|
||||
source = Path(inspect.getfile(module_class)).resolve()
|
||||
key = (str(source), module_class.__qualname__)
|
||||
if key in seen:
|
||||
continue
|
||||
seen.add(key)
|
||||
|
||||
@@ -1,299 +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 contextlib
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import tempfile
|
||||
from typing import Any, ClassVar, Dict, List, Tuple, Type, TypeVar
|
||||
|
||||
import yaml
|
||||
from appdirs import user_config_dir, user_data_dir
|
||||
from pydantic import ValidationError
|
||||
from pydantic_settings import (
|
||||
BaseSettings,
|
||||
EnvSettingsSource,
|
||||
PydanticBaseSettingsSource,
|
||||
SettingsConfigDict,
|
||||
YamlConfigSettingsSource,
|
||||
)
|
||||
|
||||
PLUGIN_CONFIG_FOLDER_NAME = "plugins"
|
||||
# Not "plugins": on macOS the configuration and data folders are the same
|
||||
# directory, and that name already holds the settings files.
|
||||
PLUGIN_DATA_FOLDER_NAME = "plugin-data"
|
||||
PLUGIN_ENV_PREFIX = "MVT_PLUGIN_"
|
||||
|
||||
PLUGIN_NAME_PATTERN = re.compile(r"[a-z0-9][a-z0-9-]*")
|
||||
|
||||
PluginSettingsType = TypeVar("PluginSettingsType", bound="MVTPluginSettings")
|
||||
|
||||
|
||||
class PluginConfigLoadError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def validate_plugin_name(plugin_name: Any) -> str:
|
||||
"""
|
||||
Check that a plugin name is safe to use in a file name and an env variable.
|
||||
|
||||
:param plugin_name: Name to validate.
|
||||
:returns: The validated plugin name.
|
||||
"""
|
||||
if not isinstance(plugin_name, str):
|
||||
raise TypeError(
|
||||
f"Plugin name must be a string, not {type(plugin_name).__name__}"
|
||||
)
|
||||
if not PLUGIN_NAME_PATTERN.fullmatch(plugin_name):
|
||||
raise ValueError(
|
||||
f"Invalid plugin name {plugin_name!r}: plugin names must start with a "
|
||||
"lowercase letter or a digit and may only contain lowercase letters, "
|
||||
"digits and dashes"
|
||||
)
|
||||
return plugin_name
|
||||
|
||||
|
||||
def plugin_config_folder() -> str:
|
||||
"""
|
||||
Return the folder where plugins store their configuration files.
|
||||
|
||||
The path is resolved on every call so it always reflects the current
|
||||
environment.
|
||||
"""
|
||||
return os.path.join(user_config_dir("mvt"), PLUGIN_CONFIG_FOLDER_NAME)
|
||||
|
||||
|
||||
def plugin_config_path(plugin_name: str) -> str:
|
||||
"""
|
||||
Return the path of the configuration file of a given plugin.
|
||||
|
||||
:param plugin_name: Name of the plugin.
|
||||
"""
|
||||
return os.path.join(
|
||||
plugin_config_folder(), f"{validate_plugin_name(plugin_name)}.yaml"
|
||||
)
|
||||
|
||||
|
||||
def plugin_data_folder(plugin_name: str) -> str:
|
||||
"""
|
||||
Return the folder where a given plugin stores its data, creating it.
|
||||
|
||||
Plugins should keep whatever they persist, such as caches or downloaded
|
||||
artifacts, in this folder. It is created with owner-only permissions. The
|
||||
path is resolved on every call so it always reflects the current
|
||||
environment. A plugin with a settings class calls
|
||||
`MVTPluginSettings.data_folder()` instead, which passes `plugin_name` here.
|
||||
|
||||
:param plugin_name: Name of the plugin.
|
||||
:returns: The path of the data folder of the plugin.
|
||||
"""
|
||||
# Validate the name before anything is created, so an unsafe name cannot
|
||||
# leave a folder behind.
|
||||
name = validate_plugin_name(plugin_name)
|
||||
|
||||
# makedirs() applies its mode only to the last folder of the path, so
|
||||
# MVT's own data folder keeps the default permissions while the two
|
||||
# plugin folders are private.
|
||||
data_folder = os.path.join(user_data_dir("mvt"), PLUGIN_DATA_FOLDER_NAME)
|
||||
os.makedirs(data_folder, mode=0o700, exist_ok=True)
|
||||
|
||||
folder = os.path.join(data_folder, name)
|
||||
os.makedirs(folder, mode=0o700, exist_ok=True)
|
||||
return folder
|
||||
|
||||
|
||||
def plugin_env_prefix(plugin_name: str) -> str:
|
||||
"""
|
||||
Return the environment variable prefix used by a given plugin.
|
||||
|
||||
Dashes are replaced by underscores. Plugin names cannot contain underscores,
|
||||
so two different plugin names never share an environment namespace.
|
||||
|
||||
:param plugin_name: Name of the plugin.
|
||||
"""
|
||||
name = validate_plugin_name(plugin_name).upper().replace("-", "_")
|
||||
return f"{PLUGIN_ENV_PREFIX}{name}_"
|
||||
|
||||
|
||||
def _settings_plugin_name(settings_cls: Type[BaseSettings]) -> str:
|
||||
plugin_name = getattr(settings_cls, "plugin_name", None)
|
||||
if plugin_name is None:
|
||||
raise TypeError(
|
||||
f"{settings_cls.__name__} must set a 'plugin_name' class attribute to "
|
||||
"namespace its configuration file and environment variables"
|
||||
)
|
||||
return validate_plugin_name(plugin_name)
|
||||
|
||||
|
||||
def _plugin_yaml_source(
|
||||
settings_cls: Type[BaseSettings], config_path: str
|
||||
) -> YamlConfigSettingsSource:
|
||||
"""
|
||||
Build the YAML settings source of a plugin, reporting unusable files.
|
||||
|
||||
A missing file is not an error, but a file which cannot be parsed or which
|
||||
does not hold a mapping of setting names is reported with its path.
|
||||
"""
|
||||
try:
|
||||
return YamlConfigSettingsSource(settings_cls, config_path)
|
||||
except yaml.YAMLError as exc:
|
||||
raise PluginConfigLoadError(
|
||||
f"Invalid plugin configuration file {config_path}: {exc}"
|
||||
) from exc
|
||||
except (TypeError, ValueError) as exc:
|
||||
raise PluginConfigLoadError(
|
||||
f"Invalid plugin configuration file {config_path}: the file must "
|
||||
"contain a mapping of setting names to values"
|
||||
) from exc
|
||||
|
||||
|
||||
class MVTPluginSettings(BaseSettings):
|
||||
"""
|
||||
Base class for plugin-namespaced settings.
|
||||
|
||||
Subclass with typed fields and set `plugin_name`. Values resolve from
|
||||
constructor arguments, then environment variables (MVT_PLUGIN_<NAME>_*),
|
||||
then the plugin's YAML file (~/.config/mvt/plugins/<name>.yaml), then field
|
||||
defaults.
|
||||
|
||||
Plugins must not store their settings in MVT's own configuration file: MVT
|
||||
rewrites it with the fields it knows about, dropping anything else.
|
||||
|
||||
`data_folder()` returns the folder the plugin keeps its data in.
|
||||
"""
|
||||
|
||||
model_config = SettingsConfigDict(extra="ignore")
|
||||
|
||||
plugin_name: ClassVar[str]
|
||||
|
||||
def __init_subclass__(cls, **kwargs: Any) -> None:
|
||||
super().__init_subclass__(**kwargs)
|
||||
plugin_name = _settings_plugin_name(cls)
|
||||
# Namespace the environment variables of this plugin. Each pydantic
|
||||
# model gets its own configuration dictionary, so this does not leak
|
||||
# into other plugins.
|
||||
cls.model_config["env_prefix"] = plugin_env_prefix(plugin_name)
|
||||
|
||||
@classmethod
|
||||
def settings_customise_sources(
|
||||
cls,
|
||||
settings_cls: Type[BaseSettings],
|
||||
init_settings: PydanticBaseSettingsSource,
|
||||
env_settings: PydanticBaseSettingsSource,
|
||||
dotenv_settings: PydanticBaseSettingsSource,
|
||||
file_secret_settings: PydanticBaseSettingsSource,
|
||||
) -> Tuple[PydanticBaseSettingsSource, ...]:
|
||||
config_path = plugin_config_path(_settings_plugin_name(settings_cls))
|
||||
yaml_source = _plugin_yaml_source(settings_cls, config_path)
|
||||
# Explicit arguments take precedence over environment variables, which
|
||||
# in turn take precedence over the configuration file.
|
||||
return (init_settings, env_settings, yaml_source)
|
||||
|
||||
@classmethod
|
||||
def load(cls: Type[PluginSettingsType]) -> PluginSettingsType:
|
||||
"""
|
||||
Load the settings of the plugin.
|
||||
|
||||
A missing configuration file is not an error: the settings then come
|
||||
from the environment and from the field defaults.
|
||||
"""
|
||||
return cls()
|
||||
|
||||
@classmethod
|
||||
def data_folder(cls) -> str:
|
||||
"""
|
||||
Return the data folder of the plugin, creating it.
|
||||
|
||||
The folder is the one plugin_data_folder() returns for `plugin_name`,
|
||||
so a plugin with a settings class does not repeat its name.
|
||||
"""
|
||||
return plugin_data_folder(_settings_plugin_name(cls))
|
||||
|
||||
def _environment_values(self) -> Dict[str, Any]:
|
||||
"""
|
||||
Return the settings values currently supplied by the environment.
|
||||
|
||||
Values are validated by the model, so they can be compared with the
|
||||
values held by this instance.
|
||||
"""
|
||||
settings_cls = type(self)
|
||||
raw_values = EnvSettingsSource(settings_cls)()
|
||||
names = [name for name in raw_values if name in settings_cls.model_fields]
|
||||
if not names:
|
||||
return {}
|
||||
|
||||
# Fall back on the current values for the fields the environment does
|
||||
# not set, so that required fields do not fail validation here.
|
||||
current_values = json.loads(self.model_dump_json())
|
||||
try:
|
||||
from_environment = settings_cls.model_validate(
|
||||
{**current_values, **raw_values}
|
||||
)
|
||||
except ValidationError:
|
||||
# An environment variable which the model cannot validate must not
|
||||
# stop the other environment values from being recognised, or a
|
||||
# credential would be written to the configuration file.
|
||||
return self._environment_values_by_field(current_values, raw_values, names)
|
||||
return {name: getattr(from_environment, name) for name in names}
|
||||
|
||||
def _environment_values_by_field(
|
||||
self,
|
||||
current_values: Dict[str, Any],
|
||||
raw_values: Dict[str, Any],
|
||||
names: List[str],
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Validate each environment value on its own, skipping unusable ones.
|
||||
|
||||
:param current_values: Serialized values held by this instance.
|
||||
:param raw_values: Values supplied by the environment.
|
||||
:param names: Names of the fields set by the environment.
|
||||
"""
|
||||
settings_cls = type(self)
|
||||
values = {}
|
||||
for name in names:
|
||||
try:
|
||||
from_environment = settings_cls.model_validate(
|
||||
{**current_values, name: raw_values[name]}
|
||||
)
|
||||
except ValidationError:
|
||||
continue
|
||||
values[name] = getattr(from_environment, name)
|
||||
return values
|
||||
|
||||
def save(self) -> None:
|
||||
"""
|
||||
Save the current settings to the configuration file of the plugin.
|
||||
|
||||
Only values which differ from the field defaults are persisted. Values
|
||||
which come from the environment are not written to disk, so credentials
|
||||
passed as environment variables stay out of the configuration file.
|
||||
MVT's own configuration file is never modified.
|
||||
"""
|
||||
config_folder = plugin_config_folder()
|
||||
if not os.path.isdir(config_folder):
|
||||
os.makedirs(config_folder, mode=0o700, exist_ok=True)
|
||||
|
||||
values = json.loads(self.model_dump_json(exclude_defaults=True))
|
||||
for name, environment_value in self._environment_values().items():
|
||||
if name in values and getattr(self, name, None) == environment_value:
|
||||
del values[name]
|
||||
|
||||
# Settings files can hold credentials, so write them through a private
|
||||
# temporary file and move it in place. The file is then never partially
|
||||
# written and never briefly readable by other users.
|
||||
config_path = plugin_config_path(self.plugin_name)
|
||||
descriptor, temporary_path = tempfile.mkstemp(
|
||||
dir=config_folder, prefix=f".{self.plugin_name}-", suffix=".yaml"
|
||||
)
|
||||
try:
|
||||
with os.fdopen(descriptor, "w") as config_file:
|
||||
config_file.write(yaml.dump(values, default_flow_style=False))
|
||||
os.replace(temporary_path, config_path)
|
||||
except BaseException:
|
||||
with contextlib.suppress(OSError):
|
||||
os.unlink(temporary_path)
|
||||
raise
|
||||
@@ -239,13 +239,6 @@ def init_logging(verbose: bool = False):
|
||||
"""
|
||||
log = logging.getLogger("mvt")
|
||||
log.setLevel(logging.DEBUG)
|
||||
|
||||
# Importing an MVT CLI module calls init_logging() at import time, and
|
||||
# loaded module packages may import one indirectly. Keep this idempotent
|
||||
# so console log lines are not duplicated by a second handler.
|
||||
if any(isinstance(handler, MVTLogHandler) for handler in log.handlers):
|
||||
return
|
||||
|
||||
consoleHandler = MVTLogHandler()
|
||||
consoleHandler.setFormatter(logging.Formatter("%(message)s"))
|
||||
if verbose:
|
||||
|
||||
@@ -1216,14 +1216,6 @@
|
||||
"version": "18.7.9",
|
||||
"build": "22H355"
|
||||
},
|
||||
{
|
||||
"version": "18.7.10",
|
||||
"build": "22H373"
|
||||
},
|
||||
{
|
||||
"version": "18.7.10",
|
||||
"build": "22H374"
|
||||
},
|
||||
{
|
||||
"version": "26",
|
||||
"build": "23A341"
|
||||
@@ -1263,13 +1255,5 @@
|
||||
{
|
||||
"version": "26.5",
|
||||
"build": "23F77"
|
||||
},
|
||||
{
|
||||
"version": "26.6.1",
|
||||
"build": "23G82"
|
||||
},
|
||||
{
|
||||
"version": "26.6.1",
|
||||
"build": "23G83"
|
||||
}
|
||||
]
|
||||
@@ -26,7 +26,6 @@ from .tcc import TCC
|
||||
from .webkit_resource_load_statistics import WebkitResourceLoadStatistics
|
||||
from .webkit_session_resource_log import WebkitSessionResourceLog
|
||||
from .whatsapp import Whatsapp
|
||||
from .whatsapp_contacts import WhatsappContacts
|
||||
|
||||
MIXED_MODULES = [
|
||||
Calls,
|
||||
@@ -48,7 +47,6 @@ MIXED_MODULES = [
|
||||
WebkitResourceLoadStatistics,
|
||||
WebkitSessionResourceLog,
|
||||
Whatsapp,
|
||||
WhatsappContacts,
|
||||
Shortcuts,
|
||||
Applications,
|
||||
Calendar,
|
||||
|
||||
@@ -3,11 +3,9 @@
|
||||
# Use of this software is governed by the MVT License 1.1 that can be found at
|
||||
# https://license.mvt.re/1.1/
|
||||
|
||||
import datetime
|
||||
import logging
|
||||
import re
|
||||
import sqlite3
|
||||
from typing import Optional, Tuple
|
||||
from typing import Optional
|
||||
|
||||
from mvt.common.module_types import (
|
||||
ModuleAtomicResult,
|
||||
@@ -17,7 +15,6 @@ from mvt.common.module_types import (
|
||||
from mvt.common.utils import convert_mactime_to_iso
|
||||
|
||||
from ..base import IOSExtraction
|
||||
from .whatsapp_contacts import WhatsappContacts
|
||||
|
||||
INTERACTIONC_BACKUP_IDS = [
|
||||
"1f5a521220a3ad80ebfdc196978df8e7a2e49dee",
|
||||
@@ -25,49 +22,6 @@ INTERACTIONC_BACKUP_IDS = [
|
||||
INTERACTIONC_ROOT_PATHS = [
|
||||
"private/var/mobile/Library/CoreDuet/People/interactionC.db",
|
||||
]
|
||||
|
||||
# The interaction record's creation date normally trails its start date by
|
||||
# milliseconds: emitting it as a timeline event only duplicates the start
|
||||
# date event. A large divergence, however, indicates the record was
|
||||
# backfilled (sync, restore, or tampering) and is worth surfacing.
|
||||
CREATION_DATE_DIVERGENCE_THRESHOLD = 3600.0
|
||||
|
||||
# Per-contact aggregate dates from ZCONTACTS are repeated on every
|
||||
# interaction row of the same contact. They are serialized with a
|
||||
# contact-centric data string so that timeline de-duplication collapses
|
||||
# them into one event per contact.
|
||||
CONTACT_EVENT_TEMPLATES = {
|
||||
"contacts_creation_date": "Contact {party} first recorded in interactionC",
|
||||
"first_incoming_sender_date": "First incoming interaction from {party}",
|
||||
"last_incoming_sender_date": "Last incoming interaction from {party}",
|
||||
"first_incoming_recipient_date": (
|
||||
"First incoming interaction where {party} was a recipient"
|
||||
),
|
||||
"last_incoming_recipient_date": (
|
||||
"Last incoming interaction where {party} was a recipient"
|
||||
),
|
||||
"first_outgoing_recipient_date": (
|
||||
"First outgoing interaction to {party}"
|
||||
),
|
||||
"last_outgoing_recipient_date": (
|
||||
"Last outgoing interaction to {party}"
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def _parse_iso(timestamp) -> Optional[datetime.datetime]:
|
||||
try:
|
||||
return datetime.datetime.strptime(
|
||||
timestamp, "%Y-%m-%d %H:%M:%S.%f"
|
||||
)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
def _describe_delta(seconds: float) -> str:
|
||||
if seconds >= 86400:
|
||||
return f"{seconds / 86400:.0f} days"
|
||||
return f"{seconds / 3600:.0f} hours"
|
||||
# Taken from APOLLO
|
||||
# https://github.com/mac4n6/APOLLO/blob/master/modules/interaction_contact_interactions.txt
|
||||
QUERIES = [
|
||||
@@ -80,7 +34,7 @@ QUERIES = [
|
||||
CASE ZINTERACTIONS.ZDIRECTION
|
||||
WHEN '0' THEN 'INCOMING'
|
||||
WHEN '1' THEN 'OUTGOING'
|
||||
END AS "direction",
|
||||
END 'DIRECTION' AS "direction",
|
||||
ZCONTACTS.ZDISPLAYNAME AS "sender_display_name",
|
||||
ZCONTACTS.ZIDENTIFIER AS "sender_identifier",
|
||||
ZCONTACTS.ZPERSONID AS "sender_personid",
|
||||
@@ -135,7 +89,7 @@ QUERIES = [
|
||||
CASE ZINTERACTIONS.ZDIRECTION
|
||||
WHEN '0' THEN 'INCOMING'
|
||||
WHEN '1' THEN 'OUTGOING'
|
||||
END AS "direction",
|
||||
END 'DIRECTION' AS "direction",
|
||||
ZCONTACTS.ZDISPLAYNAME AS "sender_display_name",
|
||||
ZCONTACTS.ZIDENTIFIER AS "sender_identifier",
|
||||
ZCONTACTS.ZPERSONID AS "sender_personid",
|
||||
@@ -163,7 +117,7 @@ QUERIES = [
|
||||
CASE ZCONTACTS.ZLASTINCOMINGRECIPIENTDATE
|
||||
WHEN '0' THEN '0'
|
||||
ELSE ZCONTACTS.ZLASTINCOMINGRECIPIENTDATE
|
||||
END AS "last_incoming_recipient_date",
|
||||
END 'LAST INCOMING RECIPIENT DATE' AS "last_incoming_recipient_date",
|
||||
ZCONTACTS.ZLASTOUTGOINGRECIPIENTDATE AS "last_outgoing_recipient_date",
|
||||
ZCONTACTS.ZCUSTOMIDENTIFIER AS "custom_id",
|
||||
ZINTERACTIONS.ZCONTENTURL AS "interaction_content_url",
|
||||
@@ -264,16 +218,9 @@ QUERIES = [
|
||||
]
|
||||
|
||||
|
||||
WHATSAPP_BUNDLE_ID = "net.whatsapp.WhatsApp"
|
||||
|
||||
|
||||
class InteractionC(IOSExtraction):
|
||||
"""This module extracts data from InteractionC db."""
|
||||
|
||||
# WhatsApp identifies chat peers by LID in interactionC.db, which only the
|
||||
# WhatsApp contacts database can map back to a phone number and name.
|
||||
dependencies = [WhatsappContacts]
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
file_path: Optional[str] = None,
|
||||
@@ -305,51 +252,10 @@ class InteractionC(IOSExtraction):
|
||||
"last_outgoing_recipient_date",
|
||||
]
|
||||
|
||||
@staticmethod
|
||||
def _describe_party(record: ModuleAtomicResult, prefix: str) -> Optional[str]:
|
||||
name = record.get(f"{prefix}_display_name") or record.get(
|
||||
f"{prefix}_resolved_name"
|
||||
)
|
||||
identifier = record.get(f"{prefix}_resolved_phone_number") or record.get(
|
||||
f"{prefix}_identifier"
|
||||
)
|
||||
if name and identifier:
|
||||
# A display name that is just a formatted copy of the phone
|
||||
# number adds no information.
|
||||
name_digits = re.sub(r"\D", "", name)
|
||||
if name_digits and name_digits == re.sub(r"\D", "", identifier):
|
||||
return identifier
|
||||
return f"{name} ({identifier})"
|
||||
return name or identifier or None
|
||||
|
||||
def serialize(self, record: ModuleAtomicResult) -> ModuleSerializedResult:
|
||||
sender = self._describe_party(record, "sender")
|
||||
# The chat peer from the domain identifier stands in when the
|
||||
# recipient was not recorded (or the recipient join is unavailable).
|
||||
recipient = self._describe_party(record, "recipient") or self._describe_party(
|
||||
record, "domain"
|
||||
)
|
||||
direction = record.get("direction")
|
||||
if not sender and direction == "OUTGOING":
|
||||
sender = "local user"
|
||||
if not recipient and direction == "INCOMING":
|
||||
recipient = "local user"
|
||||
|
||||
header = f"[{record['bundle_id']}]"
|
||||
if record.get("account"):
|
||||
header += f" {record['account']}"
|
||||
if direction:
|
||||
header += f" {direction}"
|
||||
|
||||
data = f"{header} from {sender or 'unknown'} to {recipient or 'unknown'}"
|
||||
if record.get("group_name"):
|
||||
data += f" (group: {record['group_name']})"
|
||||
if record.get("content"):
|
||||
data += f": {record['content']}"
|
||||
|
||||
records = []
|
||||
processed = []
|
||||
for timestamp in ("start_date", "end_date"):
|
||||
for timestamp in self.timestamps:
|
||||
# Check if the record has the current timestamp.
|
||||
if timestamp not in record or not record[timestamp]:
|
||||
continue
|
||||
@@ -363,142 +269,16 @@ class InteractionC(IOSExtraction):
|
||||
"timestamp": record[timestamp],
|
||||
"module": self.__class__.__name__,
|
||||
"event": timestamp,
|
||||
"data": data,
|
||||
"data": f"[{record['bundle_id']}] {record['account']} - "
|
||||
f"from {record['sender_display_name']} ({record['sender_identifier']}) "
|
||||
f"to {record.get('recipient_display_name', '')} ({record.get('recipient_identifier', '')}):"
|
||||
f" {record.get('content', '')}",
|
||||
}
|
||||
)
|
||||
processed.append(record[timestamp])
|
||||
|
||||
creation_event = self._serialize_creation_date(record, data)
|
||||
if creation_event:
|
||||
records.append(creation_event)
|
||||
|
||||
# Contact-level aggregates describe the sender's contact record.
|
||||
party = self._describe_party(record, "sender")
|
||||
if party:
|
||||
for field, template in CONTACT_EVENT_TEMPLATES.items():
|
||||
if not record.get(field):
|
||||
continue
|
||||
records.append(
|
||||
{
|
||||
"timestamp": record[field],
|
||||
"module": self.__class__.__name__,
|
||||
"event": field,
|
||||
"data": template.format(party=party),
|
||||
}
|
||||
)
|
||||
|
||||
return records
|
||||
|
||||
def _serialize_creation_date(
|
||||
self, record: ModuleAtomicResult, data: str
|
||||
) -> Optional[dict]:
|
||||
"""Serialize the interaction record's creation date only when it
|
||||
diverges from the start date enough to indicate the record was
|
||||
backfilled."""
|
||||
creation = record.get("interactions_creation_date")
|
||||
if not creation:
|
||||
return None
|
||||
|
||||
event = {
|
||||
"timestamp": creation,
|
||||
"module": self.__class__.__name__,
|
||||
"event": "interactions_creation_date",
|
||||
"data": data,
|
||||
}
|
||||
|
||||
start = _parse_iso(record.get("start_date"))
|
||||
creation_parsed = _parse_iso(creation)
|
||||
if not start or not creation_parsed:
|
||||
# Without a start date the creation date is the only anchor.
|
||||
return event
|
||||
|
||||
delta = (creation_parsed - start).total_seconds()
|
||||
if abs(delta) < CREATION_DATE_DIVERGENCE_THRESHOLD:
|
||||
return None
|
||||
|
||||
direction = "after" if delta > 0 else "before"
|
||||
event["data"] = (
|
||||
f"Interaction record created {_describe_delta(abs(delta))} "
|
||||
f"{direction} the event: {data}"
|
||||
)
|
||||
return event
|
||||
|
||||
def _whatsapp_contact_maps(self) -> Tuple[dict, dict]:
|
||||
"""Build LID and phone-digit lookup maps from the WhatsappContacts
|
||||
module results, when available."""
|
||||
by_lid: dict = {}
|
||||
by_phone: dict = {}
|
||||
contacts_module = self.dependency_modules.get(WhatsappContacts)
|
||||
if not contacts_module:
|
||||
return by_lid, by_phone
|
||||
|
||||
for contact in contacts_module.results:
|
||||
name = contact.get("full_name") or contact.get("given_name")
|
||||
phone = contact.get("phone_number")
|
||||
entry = (phone, name)
|
||||
if contact.get("lid"):
|
||||
by_lid[contact["lid"]] = entry
|
||||
if phone:
|
||||
by_phone[re.sub(r"\D", "", phone)] = entry
|
||||
whatsapp_id = contact.get("whatsapp_id")
|
||||
if whatsapp_id and "@" in whatsapp_id:
|
||||
by_phone.setdefault(whatsapp_id.split("@")[0], entry)
|
||||
|
||||
return by_lid, by_phone
|
||||
|
||||
@staticmethod
|
||||
def _resolve_whatsapp_identifier(
|
||||
value, by_lid: dict, by_phone: dict
|
||||
) -> Tuple[Optional[str], Optional[str]]:
|
||||
"""Resolve a WhatsApp identifier (LID, JID or phone number) to a
|
||||
(phone_number, contact_name) tuple."""
|
||||
if not value:
|
||||
return None, None
|
||||
|
||||
value = str(value)
|
||||
if value.endswith("@lid"):
|
||||
return by_lid.get(value, (None, None))
|
||||
if value.endswith("@g.us"):
|
||||
return None, None
|
||||
if value.endswith("@s.whatsapp.net"):
|
||||
digits = value.split("@")[0]
|
||||
phone, name = by_phone.get(digits, (None, None))
|
||||
return phone or f"+{digits}", name
|
||||
if value.startswith("+"):
|
||||
_, name = by_phone.get(re.sub(r"\D", "", value), (None, None))
|
||||
return None, name
|
||||
|
||||
return None, None
|
||||
|
||||
def _postprocess_results(self) -> None:
|
||||
by_lid, by_phone = self._whatsapp_contact_maps()
|
||||
|
||||
for entry in self.results:
|
||||
# The fallback queries return ZDIRECTION raw instead of labelled.
|
||||
if entry.get("direction") in (0, "0"):
|
||||
entry["direction"] = "INCOMING"
|
||||
elif entry.get("direction") in (1, "1"):
|
||||
entry["direction"] = "OUTGOING"
|
||||
|
||||
if entry.get("bundle_id") != WHATSAPP_BUNDLE_ID:
|
||||
continue
|
||||
|
||||
candidates = {
|
||||
"sender": [entry.get("sender_identifier"), entry.get("custom_id")],
|
||||
"recipient": [entry.get("recipient_identifier")],
|
||||
"domain": [entry.get("domain_identifier")],
|
||||
}
|
||||
for prefix, values in candidates.items():
|
||||
phone = name = None
|
||||
for value in values:
|
||||
phone, name = self._resolve_whatsapp_identifier(
|
||||
value, by_lid, by_phone
|
||||
)
|
||||
if phone or name:
|
||||
break
|
||||
entry[f"{prefix}_resolved_phone_number"] = phone
|
||||
entry[f"{prefix}_resolved_name"] = name
|
||||
|
||||
def run(self) -> None:
|
||||
self._find_ios_database(
|
||||
backup_ids=INTERACTIONC_BACKUP_IDS, root_paths=INTERACTIONC_ROOT_PATHS
|
||||
@@ -545,6 +325,4 @@ class InteractionC(IOSExtraction):
|
||||
cur.close()
|
||||
conn.close()
|
||||
|
||||
self._postprocess_results()
|
||||
|
||||
self.log.info("Extracted a total of %d InteractionC events", len(self.results))
|
||||
@@ -4,9 +4,7 @@
|
||||
# https://license.mvt.re/1.1/
|
||||
|
||||
import logging
|
||||
import os
|
||||
import sqlite3
|
||||
from typing import Dict, Optional
|
||||
from typing import Optional
|
||||
|
||||
from mvt.common.module_types import (
|
||||
ModuleAtomicResult,
|
||||
@@ -24,71 +22,9 @@ WHATSAPP_ROOT_PATHS = [
|
||||
"private/var/mobile/Containers/Shared/AppGroup/*/ChatStorage.sqlite",
|
||||
]
|
||||
|
||||
WHATSAPP_LID_BACKUP_IDS = [
|
||||
# SHA-1 of "AppDomainGroup-group.net.whatsapp.WhatsApp.shared-LID.sqlite"
|
||||
"e794f6ffcc3c222535f47684a63d5178da3c4500",
|
||||
]
|
||||
WHATSAPP_LID_ROOT_PATHS = [
|
||||
"private/var/mobile/Containers/Shared/AppGroup/*/LID.sqlite",
|
||||
]
|
||||
|
||||
# WhatsApp records the mapping between a contact's LID and phone number
|
||||
# identifiers in the ZWAPHONENUMBERLIDPAIR table. Depending on the WhatsApp
|
||||
# version this lives in a dedicated LID.sqlite database or in
|
||||
# ChatStorage.sqlite itself.
|
||||
LID_PAIRS_QUERY = """
|
||||
SELECT
|
||||
ZLID AS "lid",
|
||||
ZPHONENUMBER AS "phone_number",
|
||||
ZTIMESTAMP AS "pair_timestamp"
|
||||
FROM ZWAPHONENUMBERLIDPAIR;
|
||||
"""
|
||||
|
||||
CHAT_SESSIONS_QUERY = """
|
||||
SELECT
|
||||
ZWACHATSESSION.Z_PK AS "session_pk",
|
||||
ZWACHATSESSION.ZCONTACTJID AS "contact_jid",
|
||||
ZWACHATSESSION.ZPARTNERNAME AS "partner_name",
|
||||
ZWACHATSESSION.ZSESSIONTYPE AS "session_type",
|
||||
ZWACHATSESSION.ZARCHIVED AS "archived",
|
||||
ZWACHATSESSION.ZREMOVED AS "removed",
|
||||
ZWACHATSESSION.ZMESSAGECOUNTER AS "message_counter",
|
||||
ZWACHATSESSION.ZLASTMESSAGEDATE AS "last_message_date",
|
||||
ZWAGROUPINFO.ZCREATIONDATE AS "group_creation_date",
|
||||
MIN(ZWAMESSAGE.ZMESSAGEDATE) AS "first_stored_message_date",
|
||||
MAX(ZWAMESSAGE.ZMESSAGEDATE) AS "last_stored_message_date",
|
||||
COUNT(ZWAMESSAGE.Z_PK) AS "stored_message_count"
|
||||
FROM ZWACHATSESSION
|
||||
LEFT JOIN ZWAGROUPINFO
|
||||
ON ZWACHATSESSION.ZGROUPINFO = ZWAGROUPINFO.Z_PK
|
||||
LEFT JOIN ZWAMESSAGE
|
||||
ON ZWAMESSAGE.ZCHATSESSION = ZWACHATSESSION.Z_PK
|
||||
GROUP BY ZWACHATSESSION.Z_PK;
|
||||
"""
|
||||
|
||||
CHAT_SESSION_DATE_FIELDS = [
|
||||
"last_message_date",
|
||||
"group_creation_date",
|
||||
"first_stored_message_date",
|
||||
"last_stored_message_date",
|
||||
]
|
||||
|
||||
|
||||
def _describe_chat(record: ModuleAtomicResult) -> str:
|
||||
jid = record.get("contact_jid") or "unknown"
|
||||
name = record.get("partner_name")
|
||||
identifier = record.get("partner_resolved_phone_number") or jid
|
||||
label = f"'{name}' ({identifier})" if name else identifier
|
||||
is_group = jid.endswith("@g.us") or record.get("group_creation_date")
|
||||
if is_group:
|
||||
return f"WhatsApp group chat {label}"
|
||||
return f"WhatsApp chat with {label}"
|
||||
|
||||
|
||||
class Whatsapp(IOSExtraction):
|
||||
"""This module extracts all WhatsApp messages containing links, as well
|
||||
as per-chat records with the first and last interaction dates of each
|
||||
conversation."""
|
||||
"""This module extracts all WhatsApp messages containing links."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
@@ -109,21 +45,6 @@ class Whatsapp(IOSExtraction):
|
||||
)
|
||||
|
||||
def serialize(self, record: ModuleAtomicResult) -> ModuleSerializedResult:
|
||||
if record.get("record_type") == "chat_session":
|
||||
return self._serialize_chat_session(record)
|
||||
if record.get("record_type") == "lid_phone_number_pair":
|
||||
if not record.get("pair_timestamp"):
|
||||
return []
|
||||
return {
|
||||
"timestamp": record["pair_timestamp"],
|
||||
"module": self.__class__.__name__,
|
||||
"event": "lid_pair_recorded",
|
||||
"data": (
|
||||
f"WhatsApp associated LID {record.get('lid')} with "
|
||||
f"phone number {record.get('phone_number')}"
|
||||
),
|
||||
}
|
||||
|
||||
text = record.get("ZTEXT", "").replace("\n", "\\n")
|
||||
links_text = ""
|
||||
if record.get("links"):
|
||||
@@ -136,49 +57,6 @@ class Whatsapp(IOSExtraction):
|
||||
"data": f"'{text}' from {record.get('ZFROMJID', 'Unknown')}{links_text}",
|
||||
}
|
||||
|
||||
def _serialize_chat_session(
|
||||
self, record: ModuleAtomicResult
|
||||
) -> ModuleSerializedResult:
|
||||
records = []
|
||||
chat = _describe_chat(record)
|
||||
|
||||
if record.get("group_creation_date"):
|
||||
records.append(
|
||||
{
|
||||
"timestamp": record["group_creation_date"],
|
||||
"module": self.__class__.__name__,
|
||||
"event": "group_created",
|
||||
"data": f"{chat} was created",
|
||||
}
|
||||
)
|
||||
|
||||
if record.get("first_stored_message_date"):
|
||||
records.append(
|
||||
{
|
||||
"timestamp": record["first_stored_message_date"],
|
||||
"module": self.__class__.__name__,
|
||||
"event": "chat_first_message",
|
||||
"data": f"First stored message in {chat}",
|
||||
}
|
||||
)
|
||||
|
||||
# The chat session's own last-message date is authoritative: it can
|
||||
# postdate the newest stored message if that message was deleted.
|
||||
last_message_date = record.get("last_message_date") or record.get(
|
||||
"last_stored_message_date"
|
||||
)
|
||||
if last_message_date:
|
||||
records.append(
|
||||
{
|
||||
"timestamp": last_message_date,
|
||||
"module": self.__class__.__name__,
|
||||
"event": "chat_last_message",
|
||||
"data": f"Last message in {chat}",
|
||||
}
|
||||
)
|
||||
|
||||
return records
|
||||
|
||||
def check_indicators(self) -> None:
|
||||
if not self.indicators:
|
||||
return
|
||||
@@ -268,126 +146,7 @@ class Whatsapp(IOSExtraction):
|
||||
message["links"] = list(set(filtered_links))
|
||||
self.results.append(message)
|
||||
|
||||
total_messages = len(self.results)
|
||||
lid_map = self._extract_lid_pairs(cur)
|
||||
total_sessions = self._extract_chat_sessions(cur, lid_map)
|
||||
|
||||
cur.close()
|
||||
conn.close()
|
||||
|
||||
self.log.info(
|
||||
"Extracted a total of %d WhatsApp messages, %d chat sessions "
|
||||
"and %d LID-phone number pairs",
|
||||
total_messages,
|
||||
total_sessions,
|
||||
len(lid_map),
|
||||
)
|
||||
|
||||
def _find_lid_db_path(self) -> Optional[str]:
|
||||
for backup_id in WHATSAPP_LID_BACKUP_IDS:
|
||||
file_path = self._get_backup_file_from_id(backup_id)
|
||||
if file_path and os.path.exists(file_path):
|
||||
return file_path
|
||||
for found_path in self._get_fs_files_from_patterns(
|
||||
WHATSAPP_LID_ROOT_PATHS
|
||||
):
|
||||
return found_path
|
||||
return None
|
||||
|
||||
def _extract_lid_pairs(self, chat_cur: sqlite3.Cursor) -> Dict[str, str]:
|
||||
"""Extract the LID to phone number mapping from the dedicated
|
||||
LID.sqlite database, falling back to the same table in
|
||||
ChatStorage.sqlite. Returns a map of LID digits to phone number
|
||||
digits."""
|
||||
rows = []
|
||||
lid_db_path = self._find_lid_db_path()
|
||||
if lid_db_path:
|
||||
self.log.info(
|
||||
"Found WhatsApp LID database at path: %s", lid_db_path
|
||||
)
|
||||
lid_conn = self._open_sqlite_db(lid_db_path)
|
||||
try:
|
||||
lid_cur = lid_conn.cursor()
|
||||
lid_cur.execute(LID_PAIRS_QUERY)
|
||||
rows = lid_cur.fetchall()
|
||||
lid_cur.close()
|
||||
except sqlite3.DatabaseError as exc:
|
||||
self.log.warning(
|
||||
"Unable to extract WhatsApp LID-phone number pairs: %s",
|
||||
exc,
|
||||
)
|
||||
finally:
|
||||
lid_conn.close()
|
||||
else:
|
||||
try:
|
||||
chat_cur.execute(LID_PAIRS_QUERY)
|
||||
rows = chat_cur.fetchall()
|
||||
except sqlite3.OperationalError:
|
||||
self.log.info(
|
||||
"No WhatsApp LID database found in this backup or "
|
||||
"filesystem dump: LID chat identifiers cannot be "
|
||||
"resolved to phone numbers"
|
||||
)
|
||||
|
||||
lid_map: Dict[str, str] = {}
|
||||
for lid, phone_number, pair_timestamp in rows:
|
||||
record = {
|
||||
"record_type": "lid_phone_number_pair",
|
||||
"lid": lid,
|
||||
"phone_number": phone_number,
|
||||
"pair_timestamp": (
|
||||
convert_mactime_to_iso(pair_timestamp) or None
|
||||
)
|
||||
if pair_timestamp
|
||||
else None,
|
||||
}
|
||||
self.results.append(record)
|
||||
|
||||
if lid and phone_number:
|
||||
lid_digits = str(lid).split("@")[0]
|
||||
phone_digits = str(phone_number).split("@")[0].lstrip("+")
|
||||
lid_map[lid_digits] = phone_digits
|
||||
|
||||
return lid_map
|
||||
|
||||
def _extract_chat_sessions(
|
||||
self, cur: sqlite3.Cursor, lid_map: Dict[str, str]
|
||||
) -> int:
|
||||
"""Extract one record per chat session with the first and last
|
||||
interaction dates of each conversation."""
|
||||
try:
|
||||
cur.execute(CHAT_SESSIONS_QUERY)
|
||||
except sqlite3.OperationalError as exc:
|
||||
self.log.warning(
|
||||
"Unable to extract WhatsApp chat sessions: %s", exc
|
||||
)
|
||||
return 0
|
||||
|
||||
names = [description[0] for description in cur.description]
|
||||
|
||||
total_sessions = 0
|
||||
for row in cur.fetchall():
|
||||
session = dict(zip(names, row))
|
||||
session["record_type"] = "chat_session"
|
||||
|
||||
session["partner_resolved_phone_number"] = None
|
||||
jid = session.get("contact_jid") or ""
|
||||
if jid.endswith("@lid"):
|
||||
phone_digits = lid_map.get(jid.split("@")[0])
|
||||
if phone_digits:
|
||||
session["partner_resolved_phone_number"] = (
|
||||
f"+{phone_digits}"
|
||||
)
|
||||
|
||||
for field in CHAT_SESSION_DATE_FIELDS:
|
||||
if session.get(field):
|
||||
session[field] = (
|
||||
convert_mactime_to_iso(session[field]) or None
|
||||
)
|
||||
else:
|
||||
session[field] = None
|
||||
|
||||
self.results.append(session)
|
||||
total_sessions += 1
|
||||
|
||||
return total_sessions
|
||||
self.log.info("Extracted a total of %d WhatsApp messages", len(self.results))
|
||||
@@ -1,308 +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 logging
|
||||
import sqlite3
|
||||
from typing import Optional
|
||||
|
||||
from mvt.common.module import DatabaseNotFoundError
|
||||
from mvt.common.module_types import (
|
||||
ModuleAtomicResult,
|
||||
ModuleResults,
|
||||
ModuleSerializedResult,
|
||||
)
|
||||
from mvt.common.utils import convert_mactime_to_iso
|
||||
|
||||
from ..base import IOSExtraction
|
||||
|
||||
WHATSAPP_CONTACTS_BACKUP_IDS = [
|
||||
# SHA-1 of "AppDomainGroup-group.net.whatsapp.WhatsApp.shared-ContactsV2.sqlite"
|
||||
"b8548dc30aa1030df0ce18ef08b882cf7ab5212f",
|
||||
]
|
||||
WHATSAPP_CONTACTS_ROOT_PATHS = [
|
||||
"private/var/mobile/Containers/Shared/AppGroup/*/ContactsV2.sqlite",
|
||||
]
|
||||
|
||||
# WhatsApp's standard disappearing-messages timer values, in seconds.
|
||||
DISAPPEARING_DURATION_LABELS = {
|
||||
86400: "24 hours",
|
||||
604800: "7 days",
|
||||
1209600: "14 days",
|
||||
2592000: "30 days",
|
||||
7776000: "90 days",
|
||||
}
|
||||
|
||||
# Output field -> candidate columns in ZWAADDRESSBOOKCONTACT, in order of
|
||||
# preference. WhatsApp renames columns across versions, so the query is built
|
||||
# from the columns actually present in the database.
|
||||
COLUMN_CANDIDATES = {
|
||||
"whatsapp_id": ["ZWHATSAPPID"],
|
||||
"lid": ["ZLID"],
|
||||
"phone_number": ["ZPHONENUMBER"],
|
||||
"localized_phone_number": ["ZLOCALIZEDPHONENUMBER"],
|
||||
"full_name": ["ZFULLNAME"],
|
||||
"given_name": ["ZGIVENNAME"],
|
||||
"last_name": ["ZLASTNAME"],
|
||||
"user_name": ["ZUSERNAME"],
|
||||
"business_name": ["ZBUSINESSNAME"],
|
||||
"about_text": ["ZABOUTTEXT"],
|
||||
"about_emoji": ["ZABOUTEMOJI"],
|
||||
"notes": ["ZNOTES"],
|
||||
"disappearing_mode_duration": ["ZDISAPPEARINGMODEDURATION"],
|
||||
"disappearing_mode_timestamp": ["ZDISAPPEARINGMODETIMESTAMP"],
|
||||
"about_timestamp": ["ZABOUTTIMESTAMP"],
|
||||
"about_expiration_timestamp": ["ZABOUTEXPIRATIONTIMESTAMP"],
|
||||
"last_updated": ["ZLASTUPDATED"],
|
||||
"phone_status": ["ZPHONESTATUS", "ZPHONENUMBERSTATUS"],
|
||||
"sync_policy": ["ZSYNCPOLICY"],
|
||||
}
|
||||
|
||||
STRING_FIELDS = [
|
||||
"whatsapp_id",
|
||||
"lid",
|
||||
"phone_number",
|
||||
"localized_phone_number",
|
||||
"full_name",
|
||||
"given_name",
|
||||
"last_name",
|
||||
"user_name",
|
||||
"business_name",
|
||||
"about_text",
|
||||
"about_emoji",
|
||||
"notes",
|
||||
]
|
||||
|
||||
DATE_FIELDS = [
|
||||
"disappearing_mode_timestamp",
|
||||
"about_timestamp",
|
||||
"about_expiration_timestamp",
|
||||
"last_updated",
|
||||
]
|
||||
|
||||
|
||||
def _decode_string(value) -> Optional[str]:
|
||||
# CoreData stores string attributes as UTF-8 blobs in some WhatsApp
|
||||
# versions, so values can arrive as either bytes or str.
|
||||
if value is None:
|
||||
return None
|
||||
if isinstance(value, bytes):
|
||||
return value.decode("utf-8", "replace")
|
||||
return str(value)
|
||||
|
||||
|
||||
def _label_duration(duration) -> str:
|
||||
if not duration:
|
||||
return "off"
|
||||
return DISAPPEARING_DURATION_LABELS.get(
|
||||
int(duration), f"{int(duration)} seconds"
|
||||
)
|
||||
|
||||
|
||||
def _describe_contact(record: ModuleAtomicResult) -> str:
|
||||
contact = (
|
||||
record.get("whatsapp_id")
|
||||
or record.get("lid")
|
||||
or record.get("phone_number")
|
||||
or "unknown"
|
||||
)
|
||||
full_name = record.get("full_name")
|
||||
if full_name:
|
||||
contact = f"{contact} ({full_name})"
|
||||
return contact
|
||||
|
||||
|
||||
class WhatsappContacts(IOSExtraction):
|
||||
"""This module extracts WhatsApp contact records and per-contact
|
||||
disappearing-messages settings from ContactsV2.sqlite.
|
||||
|
||||
ChatStorage.sqlite does not record the disappearing-messages state of 1:1
|
||||
chats: the authoritative timer is stored on each contact record in this
|
||||
database, alongside the mapping between a contact's LID and phone number
|
||||
identifiers.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
file_path: Optional[str] = None,
|
||||
target_path: Optional[str] = None,
|
||||
results_path: Optional[str] = None,
|
||||
module_options: Optional[dict] = None,
|
||||
log: logging.Logger = logging.getLogger(__name__),
|
||||
results: Optional[ModuleResults] = None,
|
||||
) -> None:
|
||||
super().__init__(
|
||||
file_path=file_path,
|
||||
target_path=target_path,
|
||||
results_path=results_path,
|
||||
module_options=module_options,
|
||||
log=log,
|
||||
results=results,
|
||||
)
|
||||
|
||||
def serialize(self, record: ModuleAtomicResult) -> ModuleSerializedResult:
|
||||
records = []
|
||||
contact = _describe_contact(record)
|
||||
|
||||
if record.get("disappearing_mode_timestamp"):
|
||||
records.append(
|
||||
{
|
||||
"timestamp": record["disappearing_mode_timestamp"],
|
||||
"module": self.__class__.__name__,
|
||||
"event": "disappearing_mode_set",
|
||||
"data": (
|
||||
f"WhatsApp disappearing messages timer set to "
|
||||
f"'{record.get('disappearing_mode_label')}' "
|
||||
f"for {contact}"
|
||||
),
|
||||
}
|
||||
)
|
||||
|
||||
if record.get("about_timestamp"):
|
||||
data = f"WhatsApp about text of {contact} changed"
|
||||
about_text = record.get("about_text")
|
||||
if about_text:
|
||||
data += f' to "{about_text}"'
|
||||
records.append(
|
||||
{
|
||||
"timestamp": record["about_timestamp"],
|
||||
"module": self.__class__.__name__,
|
||||
"event": "about_changed",
|
||||
"data": data,
|
||||
}
|
||||
)
|
||||
|
||||
if record.get("about_expiration_timestamp"):
|
||||
records.append(
|
||||
{
|
||||
"timestamp": record["about_expiration_timestamp"],
|
||||
"module": self.__class__.__name__,
|
||||
"event": "about_expiration",
|
||||
"data": (
|
||||
f"WhatsApp about text of {contact} scheduled "
|
||||
f"to expire"
|
||||
),
|
||||
}
|
||||
)
|
||||
|
||||
if record.get("last_updated"):
|
||||
records.append(
|
||||
{
|
||||
"timestamp": record["last_updated"],
|
||||
"module": self.__class__.__name__,
|
||||
"event": "contact_last_updated",
|
||||
"data": f"WhatsApp contact record for {contact} updated",
|
||||
}
|
||||
)
|
||||
|
||||
return records
|
||||
|
||||
def run(self) -> None:
|
||||
try:
|
||||
self._find_ios_database(
|
||||
backup_ids=WHATSAPP_CONTACTS_BACKUP_IDS,
|
||||
root_paths=WHATSAPP_CONTACTS_ROOT_PATHS,
|
||||
)
|
||||
except DatabaseNotFoundError:
|
||||
self.log.warning(
|
||||
"Unable to find the WhatsApp ContactsV2.sqlite database in "
|
||||
"this backup or filesystem dump. WhatsApp disappearing "
|
||||
"messages settings and contact records cannot be extracted. "
|
||||
"This database is often missing from incremental backups."
|
||||
)
|
||||
return
|
||||
|
||||
self.log.info(
|
||||
"Found WhatsApp contacts database at path: %s", self.file_path
|
||||
)
|
||||
|
||||
assert self.file_path is not None
|
||||
conn = self._open_sqlite_db(self.file_path)
|
||||
cur = conn.cursor()
|
||||
try:
|
||||
try:
|
||||
cur.execute("PRAGMA table_info(ZWAADDRESSBOOKCONTACT)")
|
||||
available_columns = {row[1] for row in cur.fetchall()}
|
||||
except sqlite3.DatabaseError as exc:
|
||||
self.log.error(
|
||||
"Unable to read the ZWAADDRESSBOOKCONTACT table schema: %s",
|
||||
exc,
|
||||
)
|
||||
return
|
||||
|
||||
if not available_columns:
|
||||
self.log.warning(
|
||||
"The WhatsApp contacts database does not contain a "
|
||||
"ZWAADDRESSBOOKCONTACT table"
|
||||
)
|
||||
return
|
||||
|
||||
selected = {}
|
||||
for field, candidates in COLUMN_CANDIDATES.items():
|
||||
for candidate in candidates:
|
||||
if candidate in available_columns:
|
||||
selected[field] = candidate
|
||||
break
|
||||
|
||||
# A record with no duration column is "unknown", not "off": the
|
||||
# timer state cannot be determined from this database version.
|
||||
has_duration = "disappearing_mode_duration" in selected
|
||||
if not has_duration:
|
||||
self.log.warning(
|
||||
"The ZDISAPPEARINGMODEDURATION column is not present in "
|
||||
"this WhatsApp contacts database: disappearing messages "
|
||||
"state is unknown"
|
||||
)
|
||||
|
||||
columns = ["Z_PK"] + list(selected.values())
|
||||
cur.execute(
|
||||
f"SELECT {', '.join(columns)} FROM ZWAADDRESSBOOKCONTACT;"
|
||||
)
|
||||
fields = ["row_pk"] + list(selected.keys())
|
||||
|
||||
for row in cur:
|
||||
record = dict(zip(fields, row))
|
||||
|
||||
for field in STRING_FIELDS:
|
||||
if field in record:
|
||||
record[field] = _decode_string(record[field])
|
||||
else:
|
||||
record[field] = None
|
||||
|
||||
for field in DATE_FIELDS:
|
||||
if record.get(field) is not None:
|
||||
record[field] = (
|
||||
convert_mactime_to_iso(record[field]) or None
|
||||
)
|
||||
else:
|
||||
record[field] = None
|
||||
|
||||
duration = record.get("disappearing_mode_duration")
|
||||
if has_duration:
|
||||
record["disappearing_mode_is_on"] = bool(duration)
|
||||
record["disappearing_mode_label"] = _label_duration(
|
||||
duration
|
||||
)
|
||||
else:
|
||||
record["disappearing_mode_duration"] = None
|
||||
record["disappearing_mode_is_on"] = None
|
||||
record["disappearing_mode_label"] = None
|
||||
|
||||
record.setdefault("phone_status", None)
|
||||
record.setdefault("sync_policy", None)
|
||||
|
||||
self.results.append(record)
|
||||
finally:
|
||||
cur.close()
|
||||
conn.close()
|
||||
|
||||
total_ephemeral = sum(
|
||||
1 for record in self.results if record["disappearing_mode_is_on"]
|
||||
)
|
||||
self.log.info(
|
||||
"Extracted a total of %d WhatsApp contacts (%d with disappearing "
|
||||
"messages enabled)",
|
||||
len(self.results),
|
||||
total_ephemeral,
|
||||
)
|
||||
@@ -25,25 +25,6 @@ class TestDumpsysPackagesArtifact:
|
||||
== "com.samsung.android.provider.filterprovider"
|
||||
)
|
||||
assert dpa.results[0]["version_name"] == "5.0.07"
|
||||
assert dpa.results[0]["first_install_time"] == "2008-12-31 16:00:00"
|
||||
assert dpa.results[0]["system"] is True
|
||||
|
||||
def test_parsing_system_flag(self):
|
||||
system_details = DumpsysPackagesArtifact.parse_dumpsys_package_for_details(
|
||||
" pkgFlags=[ SYSTEM HAS_CODE ALLOW_CLEAR_USER_DATA ]"
|
||||
)
|
||||
third_party_details = DumpsysPackagesArtifact.parse_dumpsys_package_for_details(
|
||||
" pkgFlags=[ HAS_CODE ALLOW_BACKUP ]"
|
||||
)
|
||||
missing_flag_details = (
|
||||
DumpsysPackagesArtifact.parse_dumpsys_package_for_details(
|
||||
" versionName=1.0"
|
||||
)
|
||||
)
|
||||
|
||||
assert system_details["system"] is True
|
||||
assert third_party_details["system"] is False
|
||||
assert missing_flag_details["system"] is False
|
||||
|
||||
def test_ioc_check(self, indicator_file):
|
||||
dpa = DumpsysPackagesArtifact()
|
||||
@@ -59,43 +40,3 @@ class TestDumpsysPackagesArtifact:
|
||||
assert len(dpa.alertstore.alerts) == 0
|
||||
dpa.check_indicators()
|
||||
assert len(dpa.alertstore.alerts) == 1
|
||||
|
||||
def test_per_user_fields_use_primary_user(self):
|
||||
details = DumpsysPackagesArtifact.parse_dumpsys_package_for_details(
|
||||
""" User 0: installed=true
|
||||
firstInstallTime=2024-01-10 09:19:39
|
||||
runtime permissions:
|
||||
android.permission.CAMERA: granted=true
|
||||
User 95: installed=false
|
||||
firstInstallTime=1970-01-01 01:00:00
|
||||
runtime permissions:
|
||||
android.permission.CAMERA: granted=false
|
||||
android.permission.RECORD_AUDIO: granted=false
|
||||
"""
|
||||
)
|
||||
|
||||
assert details["first_install_time"] == "2024-01-10 09:19:39"
|
||||
runtime_permissions = [
|
||||
permission
|
||||
for permission in details["permissions"]
|
||||
if permission["type"] == "runtime"
|
||||
]
|
||||
assert runtime_permissions == [
|
||||
{
|
||||
"name": "android.permission.CAMERA",
|
||||
"granted": True,
|
||||
"type": "runtime",
|
||||
}
|
||||
]
|
||||
|
||||
def test_per_user_fields_fall_back_when_user_zero_is_missing(self):
|
||||
details = DumpsysPackagesArtifact.parse_dumpsys_package_for_details(
|
||||
""" User 10: installed=true
|
||||
firstInstallTime=2024-02-10 09:19:39
|
||||
runtime permissions:
|
||||
android.permission.CAMERA: granted=true
|
||||
"""
|
||||
)
|
||||
|
||||
assert details["first_install_time"] == "2024-02-10 09:19:39"
|
||||
assert details["permissions"][-1]["name"] == "android.permission.CAMERA"
|
||||
@@ -54,8 +54,6 @@ class TestBugreportAnalysis:
|
||||
== "com.samsung.android.provider.filterprovider"
|
||||
)
|
||||
assert m.results[1]["package_name"] == "com.instagram.android"
|
||||
assert m.results[0]["installer"] == ""
|
||||
assert m.results[1]["installer"] == "com.android.vending"
|
||||
assert len(m.results[0]["permissions"]) == 4
|
||||
assert len(m.results[1]["permissions"]) == 32
|
||||
|
||||
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
@@ -3,12 +3,10 @@ import pytest
|
||||
from mvt.common.module import MVTModule
|
||||
from mvt.common.module_loader import (
|
||||
CustomModuleLoadError,
|
||||
get_module_logger,
|
||||
load_custom_modules,
|
||||
load_custom_modules_from_path,
|
||||
module_supports_command,
|
||||
)
|
||||
from mvt.ios.modules.mixed.whatsapp import Whatsapp
|
||||
|
||||
|
||||
MODULE_TEMPLATE = """
|
||||
@@ -146,44 +144,3 @@ def test_module_supports_command_honors_supported_commands(tmp_path):
|
||||
|
||||
assert module_supports_command(module, "ios", "check-backup")
|
||||
assert not module_supports_command(module, "ios", "check-fs")
|
||||
|
||||
|
||||
def test_get_module_logger_keeps_builtin_names():
|
||||
assert get_module_logger(Whatsapp).name == "mvt.ios.modules.mixed.whatsapp"
|
||||
|
||||
|
||||
def test_get_module_logger_parents_package_modules_under_mvt_ext():
|
||||
class PackageModule(MVTModule):
|
||||
pass
|
||||
|
||||
PackageModule.__module__ = "some_plugin_package.ios.custom"
|
||||
|
||||
assert (
|
||||
get_module_logger(PackageModule).name
|
||||
== "mvt.ext.some_plugin_package.ios.custom"
|
||||
)
|
||||
|
||||
|
||||
def test_get_module_logger_strips_the_plugin_package_prefix():
|
||||
class PluginModule(MVTModule):
|
||||
pass
|
||||
|
||||
PluginModule.__module__ = "mvt_plugin_amnesty_custom.ios.custom"
|
||||
|
||||
assert get_module_logger(PluginModule).name == "mvt.ext.amnesty_custom.ios.custom"
|
||||
|
||||
|
||||
def test_get_module_logger_only_strips_the_prefix_from_the_top_level():
|
||||
class NestedModule(MVTModule):
|
||||
pass
|
||||
|
||||
NestedModule.__module__ = "other_package.mvt_plugin_sub"
|
||||
|
||||
assert get_module_logger(NestedModule).name == "mvt.ext.other_package.mvt_plugin_sub"
|
||||
|
||||
|
||||
def test_get_module_logger_names_path_modules_after_their_file(tmp_path):
|
||||
module_path = _write_module(tmp_path / "my_custom_module.py", "PathModule")
|
||||
module = load_custom_modules_from_path(str(module_path))[0]
|
||||
|
||||
assert get_module_logger(module).name == "mvt.ext.my_custom_module"
|
||||
@@ -1,391 +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 os
|
||||
import stat
|
||||
import sys
|
||||
from typing import Optional
|
||||
|
||||
import pytest
|
||||
import yaml
|
||||
|
||||
from mvt.common.plugin_config import (
|
||||
MVTPluginSettings,
|
||||
PluginConfigLoadError,
|
||||
plugin_config_folder,
|
||||
plugin_config_path,
|
||||
plugin_data_folder,
|
||||
plugin_env_prefix,
|
||||
)
|
||||
|
||||
|
||||
class ExamplePluginSettings(MVTPluginSettings):
|
||||
plugin_name = "example-plugin"
|
||||
|
||||
API_KEY: Optional[str] = None
|
||||
CACHE_FOLDER: str = "cache"
|
||||
MAX_RESULTS: int = 25
|
||||
|
||||
|
||||
class OtherPluginSettings(MVTPluginSettings):
|
||||
plugin_name = "other-plugin"
|
||||
|
||||
API_KEY: Optional[str] = None
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def config_folder(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.plugin_config.user_config_dir",
|
||||
lambda *args, **kwargs: str(tmp_path),
|
||||
)
|
||||
return tmp_path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def data_folder(tmp_path, monkeypatch):
|
||||
folder = tmp_path / "data"
|
||||
monkeypatch.setattr(
|
||||
"mvt.common.plugin_config.user_data_dir",
|
||||
lambda *args, **kwargs: str(folder),
|
||||
)
|
||||
return folder
|
||||
|
||||
|
||||
def _write_plugin_file(plugin_name, values):
|
||||
config_path = plugin_config_path(plugin_name)
|
||||
os.makedirs(os.path.dirname(config_path), exist_ok=True)
|
||||
content = values if isinstance(values, str) else yaml.dump(values)
|
||||
with open(config_path, "w") as config_file:
|
||||
config_file.write(content)
|
||||
return config_path
|
||||
|
||||
|
||||
def test_plugin_paths_and_prefixes_are_namespaced(config_folder):
|
||||
assert plugin_config_folder() == str(config_folder / "plugins")
|
||||
assert plugin_config_path("example-plugin") == str(
|
||||
config_folder / "plugins" / "example-plugin.yaml"
|
||||
)
|
||||
assert plugin_env_prefix("example-plugin") == "MVT_PLUGIN_EXAMPLE_PLUGIN_"
|
||||
assert plugin_env_prefix("other-plugin") == "MVT_PLUGIN_OTHER_PLUGIN_"
|
||||
|
||||
|
||||
def test_defaults_are_used_without_file_or_environment(config_folder):
|
||||
settings = ExamplePluginSettings.load()
|
||||
|
||||
assert settings.API_KEY is None
|
||||
assert settings.CACHE_FOLDER == "cache"
|
||||
assert settings.MAX_RESULTS == 25
|
||||
assert not os.path.exists(plugin_config_path("example-plugin"))
|
||||
|
||||
|
||||
def test_values_are_loaded_from_the_plugin_file(config_folder):
|
||||
_write_plugin_file("example-plugin", {"API_KEY": "from-file", "MAX_RESULTS": 5})
|
||||
|
||||
settings = ExamplePluginSettings.load()
|
||||
|
||||
assert settings.API_KEY == "from-file"
|
||||
assert settings.MAX_RESULTS == 5
|
||||
assert settings.CACHE_FOLDER == "cache"
|
||||
|
||||
|
||||
def test_environment_overrides_the_plugin_file(config_folder, monkeypatch):
|
||||
_write_plugin_file("example-plugin", {"API_KEY": "from-file", "MAX_RESULTS": 5})
|
||||
monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "from-environment")
|
||||
|
||||
settings = ExamplePluginSettings.load()
|
||||
|
||||
assert settings.API_KEY == "from-environment"
|
||||
assert settings.MAX_RESULTS == 5
|
||||
|
||||
|
||||
def test_arguments_override_the_environment_and_the_plugin_file(
|
||||
config_folder, monkeypatch
|
||||
):
|
||||
_write_plugin_file("example-plugin", {"API_KEY": "from-file", "MAX_RESULTS": 5})
|
||||
monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "from-environment")
|
||||
monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_MAX_RESULTS", "10")
|
||||
|
||||
settings = ExamplePluginSettings(API_KEY="from-argument")
|
||||
|
||||
assert settings.API_KEY == "from-argument"
|
||||
assert settings.MAX_RESULTS == 10
|
||||
|
||||
|
||||
def test_save_and_load_round_trip(config_folder):
|
||||
settings = ExamplePluginSettings.load()
|
||||
settings.API_KEY = "saved-key"
|
||||
settings.MAX_RESULTS = 100
|
||||
|
||||
settings.save()
|
||||
|
||||
config_path = plugin_config_path("example-plugin")
|
||||
assert os.path.isfile(config_path)
|
||||
with open(config_path) as config_file:
|
||||
assert yaml.safe_load(config_file) == {
|
||||
"API_KEY": "saved-key",
|
||||
"MAX_RESULTS": 100,
|
||||
}
|
||||
|
||||
reloaded = ExamplePluginSettings.load()
|
||||
assert reloaded.API_KEY == "saved-key"
|
||||
assert reloaded.MAX_RESULTS == 100
|
||||
assert reloaded.CACHE_FOLDER == "cache"
|
||||
|
||||
|
||||
@pytest.mark.skipif(
|
||||
sys.platform == "win32", reason="POSIX file permissions are not available"
|
||||
)
|
||||
def test_saved_file_is_only_readable_by_the_user(config_folder):
|
||||
settings = ExamplePluginSettings.load()
|
||||
settings.API_KEY = "saved-key"
|
||||
|
||||
settings.save()
|
||||
|
||||
config_path = plugin_config_path("example-plugin")
|
||||
assert stat.S_IMODE(os.stat(config_path).st_mode) == 0o600
|
||||
folder_mode = stat.S_IMODE(os.stat(plugin_config_folder()).st_mode)
|
||||
assert folder_mode & 0o077 == 0
|
||||
|
||||
|
||||
@pytest.mark.skipif(
|
||||
sys.platform == "win32", reason="POSIX file permissions are not available"
|
||||
)
|
||||
def test_save_restricts_the_permissions_of_an_existing_file(config_folder):
|
||||
config_path = _write_plugin_file("example-plugin", {"API_KEY": "from-file"})
|
||||
os.chmod(config_path, 0o644)
|
||||
|
||||
settings = ExamplePluginSettings.load()
|
||||
settings.MAX_RESULTS = 100
|
||||
settings.save()
|
||||
|
||||
assert stat.S_IMODE(os.stat(config_path).st_mode) == 0o600
|
||||
assert os.listdir(plugin_config_folder()) == ["example-plugin.yaml"]
|
||||
|
||||
|
||||
def test_save_only_persists_non_default_values(config_folder):
|
||||
settings = ExamplePluginSettings.load()
|
||||
settings.CACHE_FOLDER = "another-cache"
|
||||
|
||||
settings.save()
|
||||
|
||||
with open(plugin_config_path("example-plugin")) as config_file:
|
||||
assert yaml.safe_load(config_file) == {"CACHE_FOLDER": "another-cache"}
|
||||
|
||||
|
||||
def test_save_does_not_persist_values_coming_from_the_environment(
|
||||
config_folder, monkeypatch
|
||||
):
|
||||
monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "environment-secret")
|
||||
|
||||
settings = ExamplePluginSettings.load()
|
||||
assert settings.API_KEY == "environment-secret"
|
||||
settings.MAX_RESULTS = 100
|
||||
settings.save()
|
||||
|
||||
with open(plugin_config_path("example-plugin")) as config_file:
|
||||
assert yaml.safe_load(config_file) == {"MAX_RESULTS": 100}
|
||||
|
||||
|
||||
def test_an_invalid_environment_variable_still_protects_the_other_values(
|
||||
config_folder, monkeypatch
|
||||
):
|
||||
monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "environment-secret")
|
||||
monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_MAX_RESULTS", "not-an-int")
|
||||
|
||||
settings = ExamplePluginSettings(MAX_RESULTS=100)
|
||||
assert settings.API_KEY == "environment-secret"
|
||||
settings.save()
|
||||
|
||||
with open(plugin_config_path("example-plugin")) as config_file:
|
||||
saved_values = yaml.safe_load(config_file)
|
||||
assert saved_values == {"MAX_RESULTS": 100}
|
||||
assert "API_KEY" not in saved_values
|
||||
|
||||
|
||||
def test_save_persists_values_which_differ_from_the_environment(
|
||||
config_folder, monkeypatch
|
||||
):
|
||||
monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "environment-secret")
|
||||
|
||||
settings = ExamplePluginSettings.load()
|
||||
settings.API_KEY = "chosen-key"
|
||||
settings.save()
|
||||
|
||||
with open(plugin_config_path("example-plugin")) as config_file:
|
||||
assert yaml.safe_load(config_file) == {"API_KEY": "chosen-key"}
|
||||
|
||||
|
||||
def test_save_does_not_write_the_mvt_configuration_file(config_folder):
|
||||
settings = ExamplePluginSettings.load()
|
||||
settings.API_KEY = "saved-key"
|
||||
|
||||
settings.save()
|
||||
|
||||
assert os.listdir(config_folder) == ["plugins"]
|
||||
|
||||
|
||||
def test_unknown_keys_in_the_plugin_file_are_ignored(config_folder):
|
||||
_write_plugin_file(
|
||||
"example-plugin",
|
||||
{"API_KEY": "from-file", "UNKNOWN_SETTING": "ignored"},
|
||||
)
|
||||
|
||||
settings = ExamplePluginSettings.load()
|
||||
|
||||
assert settings.API_KEY == "from-file"
|
||||
assert not hasattr(settings, "UNKNOWN_SETTING")
|
||||
|
||||
|
||||
def test_unparsable_plugin_file_is_reported_with_its_path(config_folder):
|
||||
config_path = _write_plugin_file("example-plugin", "API_KEY: [unclosed\n")
|
||||
|
||||
with pytest.raises(PluginConfigLoadError) as raised:
|
||||
ExamplePluginSettings.load()
|
||||
|
||||
assert config_path in str(raised.value)
|
||||
|
||||
|
||||
def test_plugin_file_which_is_not_a_mapping_is_reported_with_its_path(config_folder):
|
||||
config_path = _write_plugin_file("example-plugin", "- one\n- two\n")
|
||||
|
||||
with pytest.raises(PluginConfigLoadError) as raised:
|
||||
ExamplePluginSettings.load()
|
||||
|
||||
assert config_path in str(raised.value)
|
||||
assert "mapping of setting names" in str(raised.value)
|
||||
|
||||
|
||||
def test_plugins_do_not_interfere_with_each_other(config_folder, monkeypatch):
|
||||
_write_plugin_file("other-plugin", {"API_KEY": "other-file-key"})
|
||||
monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "example-environment-key")
|
||||
|
||||
example_settings = ExamplePluginSettings.load()
|
||||
other_settings = OtherPluginSettings.load()
|
||||
|
||||
assert example_settings.API_KEY == "example-environment-key"
|
||||
assert other_settings.API_KEY == "other-file-key"
|
||||
|
||||
example_settings.MAX_RESULTS = 100
|
||||
example_settings.save()
|
||||
assert sorted(os.listdir(plugin_config_folder())) == [
|
||||
"example-plugin.yaml",
|
||||
"other-plugin.yaml",
|
||||
]
|
||||
with open(plugin_config_path("other-plugin")) as config_file:
|
||||
assert yaml.safe_load(config_file) == {"API_KEY": "other-file-key"}
|
||||
|
||||
|
||||
def test_subclass_without_plugin_name_is_rejected():
|
||||
with pytest.raises(TypeError, match="plugin_name"):
|
||||
|
||||
class MissingNameSettings(MVTPluginSettings):
|
||||
API_KEY: Optional[str] = None
|
||||
|
||||
|
||||
def test_subclass_with_invalid_plugin_name_is_rejected():
|
||||
with pytest.raises(ValueError, match="Invalid plugin name"):
|
||||
|
||||
class InvalidNameSettings(MVTPluginSettings):
|
||||
plugin_name = "Bad/Name"
|
||||
|
||||
|
||||
def test_underscores_are_not_allowed_in_plugin_names():
|
||||
# Underscores are replaced by dashes in the environment prefix, so allowing
|
||||
# both would let two plugin names share one environment namespace.
|
||||
with pytest.raises(ValueError, match="Invalid plugin name"):
|
||||
|
||||
class UnderscoreNameSettings(MVTPluginSettings):
|
||||
plugin_name = "under_score"
|
||||
|
||||
with pytest.raises(ValueError, match="Invalid plugin name"):
|
||||
plugin_config_path("under_score")
|
||||
with pytest.raises(ValueError, match="Invalid plugin name"):
|
||||
plugin_env_prefix("under_score")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"plugin_name", ["../escape", "folder/name", "UPPER", "-dash", ""]
|
||||
)
|
||||
def test_unsafe_plugin_names_have_no_configuration_path(plugin_name):
|
||||
with pytest.raises(ValueError, match="Invalid plugin name"):
|
||||
plugin_config_path(plugin_name)
|
||||
|
||||
|
||||
def test_data_folder_is_namespaced_and_created(data_folder):
|
||||
folder = plugin_data_folder("example-plugin")
|
||||
|
||||
assert folder == str(data_folder / "plugin-data" / "example-plugin")
|
||||
assert os.path.isdir(folder)
|
||||
|
||||
|
||||
def test_data_folder_can_be_requested_repeatedly(data_folder):
|
||||
folder = plugin_data_folder("example-plugin")
|
||||
with open(os.path.join(folder, "kept.json"), "w") as data_file:
|
||||
data_file.write("{}")
|
||||
|
||||
assert plugin_data_folder("example-plugin") == folder
|
||||
assert os.listdir(folder) == ["kept.json"]
|
||||
|
||||
|
||||
@pytest.mark.skipif(
|
||||
sys.platform == "win32", reason="POSIX file permissions are not available"
|
||||
)
|
||||
def test_data_folder_is_only_accessible_by_the_user(data_folder):
|
||||
folder = plugin_data_folder("example-plugin")
|
||||
|
||||
assert stat.S_IMODE(os.stat(folder).st_mode) & 0o077 == 0
|
||||
parent_mode = stat.S_IMODE(os.stat(os.path.dirname(folder)).st_mode)
|
||||
assert parent_mode & 0o077 == 0
|
||||
|
||||
|
||||
def test_plugins_get_their_own_data_folder(data_folder):
|
||||
example_folder = plugin_data_folder("example-plugin")
|
||||
other_folder = plugin_data_folder("other-plugin")
|
||||
|
||||
assert example_folder != other_folder
|
||||
assert sorted(os.listdir(data_folder / "plugin-data")) == [
|
||||
"example-plugin",
|
||||
"other-plugin",
|
||||
]
|
||||
|
||||
|
||||
def test_data_folder_does_not_touch_the_configuration_folder(
|
||||
config_folder, data_folder
|
||||
):
|
||||
plugin_data_folder("example-plugin")
|
||||
|
||||
assert not os.path.exists(plugin_config_folder())
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"plugin_name", ["../escape", "folder/name", "UPPER", "-dash", ""]
|
||||
)
|
||||
def test_unsafe_plugin_names_have_no_data_folder(data_folder, plugin_name):
|
||||
with pytest.raises(ValueError, match="Invalid plugin name"):
|
||||
plugin_data_folder(plugin_name)
|
||||
|
||||
assert not os.path.exists(data_folder)
|
||||
|
||||
|
||||
def test_settings_class_knows_its_data_folder(data_folder):
|
||||
folder = ExamplePluginSettings.data_folder()
|
||||
|
||||
assert folder == plugin_data_folder("example-plugin")
|
||||
assert os.path.isdir(folder)
|
||||
assert OtherPluginSettings.data_folder() != folder
|
||||
|
||||
|
||||
def test_settings_instance_uses_the_same_data_folder(config_folder, data_folder):
|
||||
settings = ExamplePluginSettings.load()
|
||||
|
||||
assert settings.data_folder() == ExamplePluginSettings.data_folder()
|
||||
|
||||
|
||||
def test_subclass_without_its_own_name_shares_the_data_folder(data_folder):
|
||||
class InheritingSettings(ExamplePluginSettings):
|
||||
pass
|
||||
|
||||
assert InheritingSettings.data_folder() == ExamplePluginSettings.data_folder()
|
||||
@@ -8,7 +8,6 @@ import logging
|
||||
import os
|
||||
from datetime import datetime
|
||||
|
||||
from mvt.common.log import MVTLogHandler
|
||||
from mvt.common.utils import (
|
||||
CustomJSONEncoder,
|
||||
convert_datetime_to_iso,
|
||||
@@ -17,7 +16,6 @@ from mvt.common.utils import (
|
||||
convert_unix_to_utc_datetime,
|
||||
generate_hashes_from_path,
|
||||
get_sha256_from_file_path,
|
||||
init_logging,
|
||||
)
|
||||
|
||||
from ..utils import get_artifact_folder
|
||||
@@ -105,20 +103,3 @@ class TestCustomJSONEncoder:
|
||||
json.dumps({"name": "家".encode()}, cls=CustomJSONEncoder)
|
||||
== '{"name": "\\u5bb6"}'
|
||||
)
|
||||
|
||||
|
||||
class TestInitLogging:
|
||||
def test__init_logging_is_idempotent(self):
|
||||
# Loaded module packages may import an MVT CLI module, which calls
|
||||
# init_logging() again at import time. A second call must not add
|
||||
# a duplicate console handler.
|
||||
log = logging.getLogger("mvt")
|
||||
init_logging()
|
||||
handler_count = sum(
|
||||
isinstance(handler, MVTLogHandler) for handler in log.handlers
|
||||
)
|
||||
init_logging()
|
||||
assert (
|
||||
sum(isinstance(handler, MVTLogHandler) for handler in log.handlers)
|
||||
== handler_count
|
||||
)
|
||||
@@ -1,105 +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/
|
||||
|
||||
from mvt.common.module import run_module
|
||||
from mvt.ios.modules.mixed.interactionc import InteractionC
|
||||
from mvt.ios.modules.mixed.whatsapp_contacts import WhatsappContacts
|
||||
|
||||
from ..utils import get_ios_backup_folder
|
||||
|
||||
|
||||
class TestInteractionCModule:
|
||||
def test_extraction_with_whatsapp_contacts(self):
|
||||
contacts = WhatsappContacts(target_path=get_ios_backup_folder())
|
||||
run_module(contacts)
|
||||
|
||||
m = InteractionC(target_path=get_ios_backup_folder())
|
||||
m.dependency_modules = {WhatsappContacts: contacts}
|
||||
run_module(m)
|
||||
|
||||
assert len(m.results) == 3
|
||||
|
||||
incoming = next(
|
||||
r for r in m.results if r["sender_identifier"] == "100000000000001@lid"
|
||||
)
|
||||
assert incoming["direction"] == "INCOMING"
|
||||
assert incoming["sender_resolved_phone_number"] == "+14155550100"
|
||||
assert incoming["sender_resolved_name"] == "Alice Example"
|
||||
|
||||
outgoing = next(
|
||||
r for r in m.results if r["direction"] == "OUTGOING"
|
||||
)
|
||||
assert outgoing["recipient_identifier"] == "+14155550100"
|
||||
assert outgoing["recipient_resolved_name"] == "Alice Example"
|
||||
assert outgoing["domain_resolved_phone_number"] == "+14155550100"
|
||||
assert outgoing["domain_resolved_name"] == "Alice Example"
|
||||
|
||||
sms = next(
|
||||
r for r in m.results if r["bundle_id"] == "com.apple.MobileSMS"
|
||||
)
|
||||
assert sms.get("sender_resolved_name") is None
|
||||
assert sms["sender_display_name"] == "Bob Example"
|
||||
|
||||
events = [entry["data"] for entry in m.timeline]
|
||||
assert (
|
||||
"[net.whatsapp.WhatsApp] INCOMING from "
|
||||
"Alice Example (+14155550100) to local user" in events
|
||||
)
|
||||
assert (
|
||||
"[net.whatsapp.WhatsApp] OUTGOING from local user to "
|
||||
"Alice Example (+14155550100)" in events
|
||||
)
|
||||
assert (
|
||||
"[com.apple.MobileSMS] INCOMING from "
|
||||
"Bob Example (+14155550101) to local user" in events
|
||||
)
|
||||
|
||||
# The creation date is only serialized when it diverges from the
|
||||
# start date; the SMS record was created 90 days after the event.
|
||||
creation_events = [
|
||||
entry
|
||||
for entry in m.timeline
|
||||
if entry["event"] == "interactions_creation_date"
|
||||
]
|
||||
assert len(creation_events) == 1
|
||||
assert creation_events[0]["timestamp"] == "2025-12-09 12:26:40.000000"
|
||||
assert creation_events[0]["data"] == (
|
||||
"Interaction record created 90 days after the event: "
|
||||
"[com.apple.MobileSMS] INCOMING from "
|
||||
"Bob Example (+14155550101) to local user"
|
||||
)
|
||||
|
||||
# Per-contact aggregate dates use contact-centric data strings.
|
||||
first_seen = [
|
||||
entry
|
||||
for entry in m.timeline
|
||||
if entry["event"] == "first_incoming_sender_date"
|
||||
]
|
||||
assert len(first_seen) == 1
|
||||
assert first_seen[0]["timestamp"] == "2025-09-03 13:46:40.000000"
|
||||
assert first_seen[0]["data"] == (
|
||||
"First incoming interaction from Bob Example (+14155550101)"
|
||||
)
|
||||
assert (
|
||||
"Last incoming interaction from Bob Example (+14155550101)"
|
||||
in events
|
||||
)
|
||||
|
||||
def test_extraction_without_whatsapp_contacts(self):
|
||||
# Without the WhatsappContacts dependency the module still runs, and
|
||||
# unresolvable LIDs are shown as-is.
|
||||
m = InteractionC(target_path=get_ios_backup_folder())
|
||||
run_module(m)
|
||||
|
||||
assert len(m.results) == 3
|
||||
events = [entry["data"] for entry in m.timeline]
|
||||
assert (
|
||||
"[net.whatsapp.WhatsApp] INCOMING from "
|
||||
"100000000000001@lid to local user" in events
|
||||
)
|
||||
assert (
|
||||
"[net.whatsapp.WhatsApp] OUTGOING from local user to "
|
||||
"+14155550100" in events
|
||||
)
|
||||
@@ -6,79 +6,8 @@
|
||||
import logging
|
||||
|
||||
from mvt.common.indicators import Indicators
|
||||
from mvt.common.module import run_module
|
||||
from mvt.ios.modules.mixed.whatsapp import Whatsapp
|
||||
|
||||
from ..utils import get_ios_backup_folder
|
||||
|
||||
|
||||
def test_extraction():
|
||||
m = Whatsapp(target_path=get_ios_backup_folder())
|
||||
run_module(m)
|
||||
|
||||
messages = [r for r in m.results if "ZTEXT" in r]
|
||||
sessions = [r for r in m.results if r.get("record_type") == "chat_session"]
|
||||
pairs = [
|
||||
r for r in m.results
|
||||
if r.get("record_type") == "lid_phone_number_pair"
|
||||
]
|
||||
assert len(messages) == 3
|
||||
assert len(sessions) == 2
|
||||
assert len(pairs) == 1
|
||||
|
||||
assert pairs[0]["lid"] == "100000000000001"
|
||||
assert pairs[0]["phone_number"] == "14155550100"
|
||||
assert pairs[0]["pair_timestamp"] == "2025-08-25 07:33:20.000000"
|
||||
|
||||
linked = next(r for r in messages if r.get("links"))
|
||||
assert linked["links"] == ["https://example.org/news"]
|
||||
|
||||
alice = next(s for s in sessions if s["partner_name"] == "Alice Example")
|
||||
assert alice["contact_jid"] == "100000000000001@lid"
|
||||
assert alice["partner_resolved_phone_number"] == "+14155550100"
|
||||
assert alice["first_stored_message_date"] == "2025-08-27 15:06:40.000000"
|
||||
assert alice["last_message_date"] == "2025-08-28 18:53:20.000000"
|
||||
assert alice["group_creation_date"] is None
|
||||
assert alice["stored_message_count"] == 2
|
||||
|
||||
group = next(s for s in sessions if s["partner_name"] == "Example Group")
|
||||
assert group["group_creation_date"] == "2025-08-21 20:13:20.000000"
|
||||
assert group["first_stored_message_date"] == "2025-08-29 22:40:00.000000"
|
||||
# The last stored message predates the session's own last-message date:
|
||||
# the newest message in this chat was deleted.
|
||||
assert group["last_stored_message_date"] == "2025-08-29 22:40:00.000000"
|
||||
assert group["last_message_date"] == "2025-08-31 02:26:40.000000"
|
||||
|
||||
# 3 message events, first/last per chat, the group creation and the
|
||||
# LID-phone number pair.
|
||||
assert len(m.timeline) == 9
|
||||
events = {
|
||||
(entry["event"], entry["timestamp"]): entry["data"]
|
||||
for entry in m.timeline
|
||||
}
|
||||
# Alice's session is keyed by LID but labelled with the phone number
|
||||
# resolved through LID.sqlite.
|
||||
assert events[("chat_first_message", "2025-08-27 15:06:40.000000")] == (
|
||||
"First stored message in WhatsApp chat with "
|
||||
"'Alice Example' (+14155550100)"
|
||||
)
|
||||
assert events[("chat_last_message", "2025-08-28 18:53:20.000000")] == (
|
||||
"Last message in WhatsApp chat with "
|
||||
"'Alice Example' (+14155550100)"
|
||||
)
|
||||
assert events[("lid_pair_recorded", "2025-08-25 07:33:20.000000")] == (
|
||||
"WhatsApp associated LID 100000000000001 with "
|
||||
"phone number 14155550100"
|
||||
)
|
||||
assert events[("group_created", "2025-08-21 20:13:20.000000")] == (
|
||||
"WhatsApp group chat 'Example Group' "
|
||||
"(120000000000000001@g.us) was created"
|
||||
)
|
||||
assert ("chat_first_message", "2025-08-29 22:40:00.000000") in events
|
||||
assert ("chat_last_message", "2025-08-31 02:26:40.000000") in events
|
||||
|
||||
assert len(m.alertstore.alerts) == 0
|
||||
|
||||
|
||||
def test_collect_url_results_includes_expansion():
|
||||
module = Whatsapp(
|
||||
|
||||
@@ -1,83 +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/
|
||||
|
||||
from mvt.common.module import run_module
|
||||
from mvt.ios.modules.mixed.whatsapp_contacts import WhatsappContacts
|
||||
|
||||
from ..utils import get_ios_backup_folder
|
||||
|
||||
|
||||
class TestWhatsappContactsModule:
|
||||
def test_extraction(self):
|
||||
m = WhatsappContacts(target_path=get_ios_backup_folder())
|
||||
run_module(m)
|
||||
assert len(m.results) == 2
|
||||
|
||||
alice = next(r for r in m.results if r["given_name"] == "Alice")
|
||||
assert alice["full_name"] == "Alice Example"
|
||||
assert alice["phone_number"] == "+14155550100"
|
||||
assert alice["whatsapp_id"] == "14155550100@s.whatsapp.net"
|
||||
assert alice["lid"] == "100000000000001@lid"
|
||||
assert alice["user_name"] == "alice.example"
|
||||
assert alice["disappearing_mode_duration"] == 86400.0
|
||||
assert alice["disappearing_mode_is_on"] is True
|
||||
assert alice["disappearing_mode_label"] == "24 hours"
|
||||
assert alice["disappearing_mode_timestamp"] == "2025-07-23 21:46:40.000000"
|
||||
assert alice["about_timestamp"] == "2025-07-12 08:00:00.000000"
|
||||
assert alice["about_expiration_timestamp"] == "2025-08-16 01:20:00.000000"
|
||||
assert alice["last_updated"] == "2025-08-04 11:33:20.000000"
|
||||
|
||||
bob = next(r for r in m.results if r["given_name"] == "Bob")
|
||||
assert bob["lid"] is None
|
||||
assert bob["disappearing_mode_duration"] is None
|
||||
assert bob["disappearing_mode_is_on"] is False
|
||||
assert bob["disappearing_mode_label"] == "off"
|
||||
assert bob["disappearing_mode_timestamp"] is None
|
||||
|
||||
# Alice: disappearing_mode_set, about_changed, about_expiration and
|
||||
# contact_last_updated. Bob: contact_last_updated only.
|
||||
assert len(m.timeline) == 5
|
||||
|
||||
events = {
|
||||
(entry["event"], entry["timestamp"]): entry["data"]
|
||||
for entry in m.timeline
|
||||
}
|
||||
assert (
|
||||
"24 hours"
|
||||
in events[("disappearing_mode_set", "2025-07-23 21:46:40.000000")]
|
||||
)
|
||||
assert (
|
||||
"14155550100@s.whatsapp.net (Alice Example)"
|
||||
in events[("disappearing_mode_set", "2025-07-23 21:46:40.000000")]
|
||||
)
|
||||
assert (
|
||||
'changed to "Hey there! I am using WhatsApp."'
|
||||
in events[("about_changed", "2025-07-12 08:00:00.000000")]
|
||||
)
|
||||
assert (
|
||||
"scheduled to expire"
|
||||
in events[("about_expiration", "2025-08-16 01:20:00.000000")]
|
||||
)
|
||||
|
||||
updated = [
|
||||
entry["data"]
|
||||
for entry in m.timeline
|
||||
if entry["event"] == "contact_last_updated"
|
||||
]
|
||||
assert len(updated) == 2
|
||||
assert all(
|
||||
entry["timestamp"] == "2025-08-04 11:33:20.000000"
|
||||
for entry in m.timeline
|
||||
if entry["event"] == "contact_last_updated"
|
||||
)
|
||||
assert any("14155550101@s.whatsapp.net (Bob Example)" in d for d in updated)
|
||||
|
||||
assert len(m.alertstore.alerts) == 0
|
||||
|
||||
def test_missing_database(self, tmp_path):
|
||||
m = WhatsappContacts(target_path=str(tmp_path))
|
||||
run_module(m)
|
||||
assert m.results == []
|
||||
assert len(m.alertstore.alerts) == 0
|
||||
@@ -15,8 +15,8 @@ class TestFilesystem:
|
||||
def test_filesystem(self):
|
||||
m = Filesystem(target_path=get_ios_backup_folder())
|
||||
run_module(m)
|
||||
assert len(m.results) == 23
|
||||
assert len(m.timeline) == 23
|
||||
assert len(m.results) == 15
|
||||
assert len(m.timeline) == 15
|
||||
assert len(m.alertstore.alerts) == 0
|
||||
|
||||
def test_detection(self, indicator_file):
|
||||
@@ -29,6 +29,6 @@ class TestFilesystem:
|
||||
)
|
||||
m.indicators = ind
|
||||
run_module(m)
|
||||
assert len(m.results) == 23
|
||||
assert len(m.timeline) == 23
|
||||
assert len(m.results) == 15
|
||||
assert len(m.timeline) == 15
|
||||
assert len(m.alertstore.alerts) == 1
|
||||
@@ -1,7 +1,3 @@
|
||||
import hashlib
|
||||
import importlib.metadata
|
||||
import json
|
||||
|
||||
from click.testing import CliRunner
|
||||
|
||||
from mvt.android.cli import check_bugreport
|
||||
@@ -9,9 +5,7 @@ from mvt.android.cmd_check_androidqf import CmdAndroidCheckAndroidQF
|
||||
from mvt.android.cmd_check_backup import CmdAndroidCheckBackup
|
||||
from mvt.android.cmd_check_bugreport import CmdAndroidCheckBugreport
|
||||
from mvt.android.cmd_check_intrusion_logs import CmdAndroidCheckIntrusionLogs
|
||||
from mvt.common import module_loader
|
||||
from mvt.common.module import MVTModule
|
||||
from mvt.common.version import MVT_VERSION
|
||||
from mvt.ios.cli import check_backup, check_fs
|
||||
|
||||
|
||||
@@ -112,150 +106,6 @@ def test_custom_modules_load_from_environment_without_cli_flag(tmp_path, monkeyp
|
||||
assert "EnvBugreportModule" in result.output
|
||||
|
||||
|
||||
class InstalledPackageModule(MVTModule):
|
||||
supported_commands = (("ios", "check-backup"),)
|
||||
|
||||
|
||||
def get_installed_package_modules():
|
||||
return [InstalledPackageModule]
|
||||
|
||||
|
||||
def _fake_entry_points(monkeypatch, value, name="test-modules"):
|
||||
entry_point = importlib.metadata.EntryPoint(
|
||||
name=name, value=value, group=module_loader.MODULES_ENTRY_POINT_GROUP
|
||||
)
|
||||
|
||||
def fake_entry_points(*, group):
|
||||
assert group == module_loader.MODULES_ENTRY_POINT_GROUP
|
||||
return [entry_point]
|
||||
|
||||
monkeypatch.setattr(
|
||||
module_loader.importlib.metadata, "entry_points", fake_entry_points
|
||||
)
|
||||
|
||||
|
||||
def test_installed_module_package_loads_from_entry_point(monkeypatch):
|
||||
_fake_entry_points(monkeypatch, f"{__name__}:get_installed_package_modules")
|
||||
|
||||
modules = module_loader.load_custom_modules()
|
||||
|
||||
assert modules == [InstalledPackageModule]
|
||||
|
||||
|
||||
def test_broken_module_entry_point_is_skipped(monkeypatch, caplog):
|
||||
_fake_entry_points(monkeypatch, "nonexistent_module_xyz:get_modules")
|
||||
|
||||
with caplog.at_level("WARNING"):
|
||||
modules = module_loader.load_custom_modules()
|
||||
|
||||
assert modules == []
|
||||
assert "Unable to load modules from entry point" in caplog.text
|
||||
|
||||
|
||||
def test_entry_point_module_deduplicated_against_paths(monkeypatch, tmp_path):
|
||||
_fake_entry_points(monkeypatch, f"{__name__}:get_installed_package_modules")
|
||||
module_path = _write_custom_module(
|
||||
tmp_path / "custom.py",
|
||||
"PathLoadedModule",
|
||||
(("ios", "check-backup"),),
|
||||
)
|
||||
|
||||
modules = module_loader.load_custom_modules([str(module_path)])
|
||||
|
||||
assert [module.__name__ for module in modules] == [
|
||||
"InstalledPackageModule",
|
||||
"PathLoadedModule",
|
||||
]
|
||||
|
||||
|
||||
def test_list_modules_shows_module_sources(tmp_path, caplog):
|
||||
module_path = _write_custom_module(
|
||||
tmp_path / "custom.py",
|
||||
"SourcedBackupModule",
|
||||
(("ios", "check-backup"),),
|
||||
)
|
||||
file_sha256 = hashlib.sha256(module_path.read_bytes()).hexdigest()
|
||||
custom_modules = module_loader.load_custom_modules([str(module_path)])
|
||||
|
||||
from mvt.ios.cmd_check_backup import CmdIOSCheckBackup
|
||||
|
||||
cmd = CmdIOSCheckBackup(target_path=str(tmp_path), custom_modules=custom_modules)
|
||||
cmd.list_modules()
|
||||
|
||||
assert f" - Modules from 'mvt@{MVT_VERSION}':" in caplog.text
|
||||
assert (
|
||||
f" - Modules from '{module_path}' (sha256: {file_sha256}): SourcedBackupModule"
|
||||
in caplog.text
|
||||
)
|
||||
|
||||
|
||||
def test_builtin_module_origin():
|
||||
from mvt.ios.modules.backup import BACKUP_MODULES
|
||||
|
||||
origin = module_loader.get_module_origin(BACKUP_MODULES[0])
|
||||
|
||||
assert origin.kind == "builtin"
|
||||
assert origin.name == "mvt"
|
||||
assert origin.version == MVT_VERSION
|
||||
|
||||
|
||||
def test_installed_module_origin(monkeypatch):
|
||||
_fake_entry_points(monkeypatch, f"{__name__}:get_installed_package_modules")
|
||||
|
||||
modules = module_loader.load_custom_modules()
|
||||
|
||||
origin = module_loader.get_module_origin(modules[0])
|
||||
assert origin.kind == "package"
|
||||
assert origin.name == "test-modules"
|
||||
|
||||
|
||||
def test_distribution_commit_read_from_direct_url():
|
||||
class FakeDistribution:
|
||||
def read_text(self, filename):
|
||||
assert filename == "direct_url.json"
|
||||
return json.dumps(
|
||||
{
|
||||
"url": "https://github.com/example/example-modules",
|
||||
"vcs_info": {"commit_id": "abc1234", "vcs": "git"},
|
||||
}
|
||||
)
|
||||
|
||||
assert module_loader._distribution_commit(FakeDistribution()) == "abc1234"
|
||||
|
||||
|
||||
def test_command_log_records_loaded_modules(tmp_path):
|
||||
(tmp_path / "Manifest.db").touch()
|
||||
(tmp_path / "Info.plist").touch()
|
||||
module_path = _write_custom_module(
|
||||
tmp_path / "custom.py",
|
||||
"AuditedRunModule",
|
||||
(("ios", "check-backup"),),
|
||||
slug="audited_run_module",
|
||||
)
|
||||
file_sha256 = hashlib.sha256(module_path.read_bytes()).hexdigest()
|
||||
output_path = tmp_path / "out"
|
||||
|
||||
result = CliRunner().invoke(
|
||||
check_backup,
|
||||
[
|
||||
"--module",
|
||||
"AuditedRunModule",
|
||||
"--load-module",
|
||||
str(module_path),
|
||||
"--output",
|
||||
str(output_path),
|
||||
str(tmp_path),
|
||||
],
|
||||
)
|
||||
|
||||
assert result.exit_code == 0
|
||||
command_log = (output_path / "command.log").read_text(encoding="utf-8")
|
||||
assert (
|
||||
f"Loaded 1 check-backup modules from '{module_path}' "
|
||||
f"(sha256: {file_sha256}): AuditedRunModule" in command_log
|
||||
)
|
||||
|
||||
|
||||
class NestedBugreportModule(MVTModule):
|
||||
supported_commands = (("android", "check-bugreport"),)
|
||||
|
||||
|
||||
Reference in new issue
Block a user