mirror of
https://github.com/mvt-project/mvt.git
synced 2026-10-25 23:53:42 +01:00
Compare commits
6
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f5b7045b03 | ||
|
|
8617e0bf54 | ||
|
|
c261316d7a | ||
|
|
53fb12aee8 | ||
|
|
8140e350f7 | ||
|
|
f483223e23 |
No files matched your search
@@ -57,3 +57,9 @@ If the backup is encrypted, ABE will prompt you to enter the password.
|
||||
Alternatively, [ab-decrypt](https://github.com/joernheissler/ab-decrypt) can be used for that purpose.
|
||||
|
||||
You can then extract SMSs with MVT by passing the folder path as parameter instead of the `.ab` file: `mvt-android check-backup --output /path/to/results/ /path/to/backup/` (the path to backup given should be the folder containing the `apps` folder).
|
||||
|
||||
When an output folder is specified, URLs extracted from SMS and MMS messages
|
||||
are also written to `urls.json`. Each entry contains the URL, its expanded
|
||||
destination when MVT resolved a shortened URL during indicator checking, the
|
||||
message timestamp, and the `sms` source. The same file is created by
|
||||
`check-androidqf` when its nested Android backup contains messages with URLs.
|
||||
+5
-3
@@ -82,14 +82,16 @@ class ExampleCustomModule(MVTModule):
|
||||
return None
|
||||
```
|
||||
|
||||
Use `supported_commands` to restrict a module to specific platform/command
|
||||
pairs. Missing or empty `supported_commands` means the module is available to
|
||||
all commands, which keeps older modules compatible. Supported pairs are:
|
||||
Use `supported_commands` to declare the platform/command pairs a module
|
||||
supports. Empty `supported_commands` means the module will not run and MVT logs
|
||||
a warning. This explicit declaration is required for every command. Supported
|
||||
pairs are:
|
||||
|
||||
```python
|
||||
("ios", "check-backup")
|
||||
("ios", "check-fs")
|
||||
("ios", "check-iocs")
|
||||
("ios", "check-sysdiagnose")
|
||||
("android", "check-backup")
|
||||
("android", "check-bugreport")
|
||||
("android", "check-androidqf")
|
||||
|
||||
@@ -312,6 +312,15 @@ If indicators are provided through the command-line, they are checked against th
|
||||
|
||||
---
|
||||
|
||||
### `urls.json`
|
||||
|
||||
This JSON file collects URLs extracted from SMS, iMessage, and WhatsApp
|
||||
messages. Each entry contains the original URL, its expanded destination when
|
||||
MVT resolved a shortened URL during indicator checking, the message timestamp,
|
||||
and its `sms` or `whatsapp` source.
|
||||
|
||||
---
|
||||
|
||||
### `sms_attachments.json`
|
||||
|
||||
!!! info "Availability"
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
# Check an iOS Sysdiagnose
|
||||
|
||||
`mvt-ios check-sysdiagnose` prepares an iOS sysdiagnose archive for analysis by
|
||||
custom MVT modules. MVT does not include built-in sysdiagnose modules. You must
|
||||
load at least one custom module that explicitly supports this command.
|
||||
|
||||
The command accepts either an extracted sysdiagnose directory or the original
|
||||
gzip-compressed tar archive.
|
||||
|
||||
```bash
|
||||
mvt-ios check-sysdiagnose \
|
||||
--load-module ./sysdiagnose_modules.py \
|
||||
--output ./results \
|
||||
./sysdiagnose_2024.01.02_03-04-05+0200.tar.gz
|
||||
```
|
||||
|
||||
Use `--hashes` to include hashes for analyzed files in `info.json`, and
|
||||
`--list-modules` to display the eligible custom modules without running them.
|
||||
|
||||
## Writing a custom module
|
||||
|
||||
Extend `SysdiagnoseExtraction` to access the archive contents consistently for
|
||||
both directory and tar inputs. Each module must declare the command explicitly
|
||||
in `supported_commands`.
|
||||
|
||||
```python
|
||||
from mvt.ios.modules.sysdiagnose import SysdiagnoseExtraction
|
||||
|
||||
|
||||
class ExampleSysdiagnoseModule(SysdiagnoseExtraction):
|
||||
supported_commands = (("ios", "check-sysdiagnose"),)
|
||||
slug = "example_sysdiagnose"
|
||||
|
||||
def run(self):
|
||||
paths = self._get_files_by_pattern("*/example.log")
|
||||
if paths:
|
||||
content = self._get_file_content(paths[0]).decode("utf-8", "replace")
|
||||
self.results = [{"content": content}]
|
||||
|
||||
def check_indicators(self):
|
||||
pass
|
||||
|
||||
def serialize(self, result):
|
||||
return None
|
||||
```
|
||||
|
||||
The base class provides `from_sysdiagnose_folder()` and
|
||||
`from_sysdiagnose_tar()` setup hooks, as well as protected file lookup, file
|
||||
reading, and timezone extraction helpers. IPS crash-report metadata is exposed
|
||||
on `ips_files`.
|
||||
@@ -3,6 +3,8 @@ repo_url: https://github.com/mvt-project/mvt
|
||||
edit_uri: edit/main/docs/
|
||||
copyright: Copyright © 2021-2023 MVT Project Developers
|
||||
site_description: Mobile Verification Toolkit Documentation
|
||||
not_in_nav: |
|
||||
android/adb.md
|
||||
markdown_extensions:
|
||||
- attr_list
|
||||
- admonition
|
||||
@@ -28,6 +30,7 @@ nav:
|
||||
- Welcome: "index.md"
|
||||
- Introduction: "introduction.md"
|
||||
- Installation: "install.md"
|
||||
- Command Completion: "command_completion.md"
|
||||
- Using Docker: "docker.md"
|
||||
- MVT for iOS:
|
||||
- iOS Forensic Methodology: "ios/methodology.md"
|
||||
@@ -39,6 +42,7 @@ nav:
|
||||
- Check a Filesystem Dump:
|
||||
- Dumping the filesystem: "ios/filesystem/dump.md"
|
||||
- Check a Filesystem Dump with mvt-ios: "ios/filesystem/check.md"
|
||||
- Check a Sysdiagnose: "ios/sysdiagnose.md"
|
||||
- Records extracted by mvt-ios: "ios/records.md"
|
||||
- MVT for Android:
|
||||
- Android Forensic Methodology: "android/methodology.md"
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
# Use of this software is governed by the MVT License 1.1 that can be found at
|
||||
# https://license.mvt.re/1.1/
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
@@ -94,6 +95,47 @@ class CmdAndroidCheckAndroidQF(Command):
|
||||
self.__zip = zipfile.ZipFile(self.target_path)
|
||||
self.__files = self.__zip.namelist()
|
||||
|
||||
self._load_acquisition_context()
|
||||
|
||||
def _load_acquisition_context(self) -> None:
|
||||
"""Pass AndroidQF acquisition metadata to nested commands and modules."""
|
||||
context = {}
|
||||
metadata_files = [
|
||||
file_path
|
||||
for file_path in self.__files
|
||||
if file_path.replace("\\", "/").rsplit("/", 1)[-1] == "acquisition.json"
|
||||
]
|
||||
for file_path in metadata_files:
|
||||
try:
|
||||
metadata = json.loads(self._get_file_content(file_path))
|
||||
if isinstance(metadata, dict):
|
||||
context["started"] = metadata.get("started")
|
||||
context["adb_host_public_key"] = metadata.get("adb_host_public_key")
|
||||
break
|
||||
except (json.JSONDecodeError, OSError, TypeError, UnicodeDecodeError):
|
||||
self.log.warning(
|
||||
'Unable to read AndroidQF acquisition metadata "%s"', file_path
|
||||
)
|
||||
|
||||
if not context.get("adb_host_public_key"):
|
||||
key_files = [
|
||||
file_path
|
||||
for file_path in self.__files
|
||||
if file_path.replace("\\", "/").rsplit("/", 1)[-1] == "adb_host_key.pub"
|
||||
]
|
||||
for file_path in key_files:
|
||||
try:
|
||||
context["adb_host_public_key"] = self._get_file_content(
|
||||
file_path
|
||||
).decode("utf-8")
|
||||
break
|
||||
except (OSError, UnicodeDecodeError):
|
||||
self.log.warning(
|
||||
'Unable to read AndroidQF ADB host key "%s"', file_path
|
||||
)
|
||||
|
||||
self.module_options["androidqf_acquisition"] = context
|
||||
|
||||
def module_init(self, module: AndroidQFModule) -> None: # type: ignore[override]
|
||||
if self.__format == "zip" and self.__zip:
|
||||
module.from_zip(self.__zip, self.__files)
|
||||
@@ -220,6 +262,7 @@ class CmdAndroidCheckAndroidQF(Command):
|
||||
cmd.run()
|
||||
|
||||
self.timeline.extend(cmd.timeline)
|
||||
self.url_results.extend(cmd.url_results)
|
||||
self.alertstore.extend(cmd.alertstore.alerts)
|
||||
finally:
|
||||
if bugreport:
|
||||
@@ -257,6 +300,7 @@ class CmdAndroidCheckAndroidQF(Command):
|
||||
cmd.run()
|
||||
|
||||
self.timeline.extend(cmd.timeline)
|
||||
self.url_results.extend(cmd.url_results)
|
||||
self.alertstore.extend(cmd.alertstore.alerts)
|
||||
return True
|
||||
|
||||
@@ -336,6 +380,7 @@ class CmdAndroidCheckAndroidQF(Command):
|
||||
cmd.run()
|
||||
|
||||
self.timeline.extend(cmd.timeline)
|
||||
self.url_results.extend(cmd.url_results)
|
||||
self.alertstore.extend(cmd.alertstore.alerts)
|
||||
return True
|
||||
|
||||
|
||||
@@ -57,6 +57,11 @@ class SMS(BackupModule):
|
||||
ioc_match.message, "", message, matched_indicator=ioc_match.ioc
|
||||
)
|
||||
|
||||
def collect_url_results(self) -> None:
|
||||
for message in self.results:
|
||||
for url in message.get("links", []):
|
||||
self.add_url_result(url, message.get("isodate"), "sms")
|
||||
|
||||
def run(self) -> None:
|
||||
sms_path = "apps/com.android.providers.telephony/d_f/*_sms_backup"
|
||||
for file in self._get_files_by_pattern(sms_path):
|
||||
|
||||
@@ -3,11 +3,15 @@
|
||||
# Use of this software is governed by the MVT License 1.1 that can be found at
|
||||
# https://license.mvt.re/1.1/
|
||||
|
||||
import base64
|
||||
import binascii
|
||||
import datetime
|
||||
import logging
|
||||
from typing import Optional
|
||||
|
||||
from mvt.android.artifacts.dumpsys_adb import DumpsysADBArtifact
|
||||
from mvt.common.module_types import ModuleResults
|
||||
from mvt.common.utils import convert_datetime_to_iso
|
||||
|
||||
from .base import BugReportModule
|
||||
|
||||
@@ -53,3 +57,122 @@ class DumpsysADBState(DumpsysADBArtifact, BugReportModule):
|
||||
"Identified a total of %d trusted ADB keys",
|
||||
len(self.results[0].get("user_keys", [])),
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _key_material(public_key: object) -> str:
|
||||
if isinstance(public_key, bytes):
|
||||
public_key = public_key.decode("utf-8", errors="replace")
|
||||
if not isinstance(public_key, str):
|
||||
return ""
|
||||
return public_key.strip().split(" ", 1)[0]
|
||||
|
||||
@staticmethod
|
||||
def _is_valid_key(public_key: str) -> bool:
|
||||
if not public_key:
|
||||
return False
|
||||
try:
|
||||
return bool(base64.b64decode(public_key, validate=True))
|
||||
except (binascii.Error, ValueError):
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def _parse_acquisition_time(value: object) -> Optional[datetime.datetime]:
|
||||
if not isinstance(value, str):
|
||||
return None
|
||||
try:
|
||||
timestamp = datetime.datetime.fromisoformat(value.replace("Z", "+00:00"))
|
||||
except ValueError:
|
||||
return None
|
||||
if not timestamp.tzinfo:
|
||||
timestamp = timestamp.replace(tzinfo=datetime.timezone.utc)
|
||||
return timestamp.astimezone(datetime.timezone.utc)
|
||||
|
||||
@staticmethod
|
||||
def _parse_last_connected(value: object) -> Optional[datetime.datetime]:
|
||||
try:
|
||||
return datetime.datetime.fromtimestamp(
|
||||
int(str(value)) / 1000,
|
||||
tz=datetime.timezone.utc,
|
||||
)
|
||||
except (OSError, OverflowError, TypeError, ValueError):
|
||||
return None
|
||||
|
||||
def _trusted_keys(self) -> list[dict]:
|
||||
"""Return unique trusted keys, preferring keystore connection metadata."""
|
||||
trusted_keys = []
|
||||
seen = set()
|
||||
for result in self.results:
|
||||
keystore = result.get("keystore", [])
|
||||
candidates = keystore if isinstance(keystore, list) else []
|
||||
candidates = [*candidates, *result.get("user_keys", [])]
|
||||
for candidate in candidates:
|
||||
if not isinstance(candidate, dict):
|
||||
continue
|
||||
key = self._key_material(candidate.get("key"))
|
||||
identity = key or repr(candidate)
|
||||
if identity in seen:
|
||||
continue
|
||||
seen.add(identity)
|
||||
trusted_keys.append(candidate)
|
||||
return trusted_keys
|
||||
|
||||
def check_indicators(self) -> None:
|
||||
if "androidqf_acquisition" not in self.module_options:
|
||||
return super().check_indicators()
|
||||
|
||||
context = self.module_options.get("androidqf_acquisition")
|
||||
if not isinstance(context, dict):
|
||||
context = {}
|
||||
acquisition_key = self._key_material(context.get("adb_host_public_key"))
|
||||
if acquisition_key and not self._is_valid_key(acquisition_key):
|
||||
acquisition_key = ""
|
||||
acquisition_time = self._parse_acquisition_time(context.get("started"))
|
||||
cutoff = (
|
||||
acquisition_time - datetime.timedelta(days=1) if acquisition_time else None
|
||||
)
|
||||
|
||||
for trusted_key in self._trusted_keys():
|
||||
key = self._key_material(trusted_key.get("key"))
|
||||
fingerprint = trusted_key.get("fingerprint") or "<unknown key>"
|
||||
user = trusted_key.get("user") or "unknown user"
|
||||
description = f"{fingerprint} ({user})"
|
||||
last_connected = self._parse_last_connected(
|
||||
trusted_key.get("last_connected")
|
||||
)
|
||||
event_time = (
|
||||
convert_datetime_to_iso(last_connected) if last_connected else ""
|
||||
)
|
||||
|
||||
if not self._is_valid_key(key):
|
||||
self.alertstore.low(
|
||||
f"Found an invalid trusted ADB host key: {description}",
|
||||
event_time,
|
||||
trusted_key,
|
||||
)
|
||||
continue
|
||||
|
||||
if not acquisition_key:
|
||||
self.alertstore.low(
|
||||
"Found a trusted ADB host key, but the AndroidQF acquisition "
|
||||
f"does not include its host key: {description}",
|
||||
event_time,
|
||||
trusted_key,
|
||||
)
|
||||
continue
|
||||
|
||||
if key != acquisition_key:
|
||||
self.alertstore.low(
|
||||
"Found a trusted ADB host key different from the AndroidQF "
|
||||
f"acquisition host: {description}",
|
||||
event_time,
|
||||
trusted_key,
|
||||
)
|
||||
continue
|
||||
|
||||
if cutoff and last_connected and last_connected <= cutoff:
|
||||
self.alertstore.info(
|
||||
"Found a trusted ADB host key last connected at least one day "
|
||||
f"before the AndroidQF acquisition: {description}",
|
||||
event_time,
|
||||
trusted_key,
|
||||
)
|
||||
@@ -20,7 +20,7 @@ from .config import settings
|
||||
from .indicators import Indicators
|
||||
from .module import EncryptedBackupError, MVTModule, run_module, save_timeline
|
||||
from .module_loader import module_supports_command
|
||||
from .module_types import ModuleTimeline
|
||||
from .module_types import ModuleTimeline, URLResult
|
||||
from .utils import (
|
||||
CustomJSONEncoder,
|
||||
convert_datetime_to_iso,
|
||||
@@ -73,6 +73,7 @@ class Command:
|
||||
self.hashes = hashes
|
||||
self.hash_values: list[dict[str, Any]] = []
|
||||
self.timeline: ModuleTimeline = []
|
||||
self.url_results: list[URLResult] = []
|
||||
|
||||
# Load IOCs
|
||||
self._create_storage()
|
||||
@@ -150,6 +151,14 @@ class Command:
|
||||
with open(alerts_path, "w+", encoding="utf-8") as handle:
|
||||
json.dump(alerts, handle, indent=4, cls=CustomJSONEncoder)
|
||||
|
||||
def _store_urls(self) -> None:
|
||||
if not self.results_path or not self.url_results:
|
||||
return
|
||||
|
||||
urls_path = os.path.join(self.results_path, "urls.json")
|
||||
with open(urls_path, "w", encoding="utf-8") as handle:
|
||||
json.dump(self.url_results, handle, indent=4, cls=CustomJSONEncoder)
|
||||
|
||||
def _store_alerts_timeline(self) -> None:
|
||||
if not self.results_path:
|
||||
return
|
||||
@@ -396,6 +405,7 @@ class Command:
|
||||
self.executed.append(m)
|
||||
executed_by_type[module] = m
|
||||
self.timeline.extend(m.timeline)
|
||||
self.url_results.extend(m.url_results)
|
||||
self.alertstore.extend(m.alertstore.alerts)
|
||||
|
||||
try:
|
||||
@@ -410,4 +420,5 @@ class Command:
|
||||
self._store_timeline()
|
||||
self._store_alerts_timeline()
|
||||
self._store_alerts()
|
||||
self._store_urls()
|
||||
self._store_info()
|
||||
@@ -38,6 +38,7 @@ HELP_MSG_BACKUP_KEYFILE = (
|
||||
HELP_MSG_EXTRACT_KEY = "Extract decryption key from an iTunes backup"
|
||||
HELP_MSG_CHECK_IOS_BACKUP = "Extract artifacts from an iTunes backup"
|
||||
HELP_MSG_CHECK_FS = "Extract artifacts from a full filesystem dump"
|
||||
HELP_MSG_CHECK_SYSDIAGNOSE = "Extract artifacts from an iOS sysdiagnose archive"
|
||||
|
||||
# Android Specific
|
||||
HELP_MSG_ANDROID_BACKUP_PASSWORD = "The backup password to use for an Android backup"
|
||||
|
||||
@@ -49,6 +49,7 @@ class Indicators:
|
||||
self.log = log
|
||||
self.ioc_collections: List[Dict[str, Any]] = []
|
||||
self.total_ioc_count = 0
|
||||
self.resolved_urls: Dict[str, str] = {}
|
||||
|
||||
def _load_downloaded_indicators(self) -> None:
|
||||
if not os.path.isdir(MVT_INDICATORS_FOLDER):
|
||||
@@ -439,9 +440,14 @@ class Indicators:
|
||||
orig_url.url,
|
||||
dest_url.url,
|
||||
)
|
||||
return self.check_url(dest_url.url)
|
||||
match = self.check_url(dest_url.url)
|
||||
self.resolved_urls[url] = self.resolved_urls.get(
|
||||
dest_url.url, dest_url.url
|
||||
)
|
||||
return match
|
||||
|
||||
final_url = dest_url
|
||||
self.resolved_urls[url] = final_url.url
|
||||
else:
|
||||
# If it's not shortened, we just use the original URL object.
|
||||
final_url = orig_url
|
||||
@@ -482,6 +488,13 @@ class Indicators:
|
||||
|
||||
return None
|
||||
|
||||
def get_expanded_url(self, url: str) -> Optional[str]:
|
||||
"""Return the final URL recorded while checking a shortened URL."""
|
||||
expanded_url = self.resolved_urls.get(url)
|
||||
if expanded_url and expanded_url != url:
|
||||
return expanded_url
|
||||
return None
|
||||
|
||||
def check_urls(self, urls: list) -> Optional[IndicatorMatch]:
|
||||
"""Check a list of URLs against the provided list of domain indicators.
|
||||
|
||||
|
||||
@@ -18,6 +18,7 @@ from .module_types import (
|
||||
ModuleResults,
|
||||
ModuleSerializedResult,
|
||||
ModuleTimeline,
|
||||
URLResult,
|
||||
)
|
||||
from .utils import CustomJSONEncoder, exec_or_profile
|
||||
|
||||
@@ -82,6 +83,7 @@ class MVTModule:
|
||||
|
||||
self.results: ModuleResults = results if results is not None else []
|
||||
self.timeline: ModuleTimeline = []
|
||||
self.url_results: list[URLResult] = []
|
||||
self.dependency_modules: Dict[type["MVTModule"], "MVTModule"] = {}
|
||||
|
||||
def get_dependency_results(
|
||||
@@ -110,6 +112,23 @@ class MVTModule:
|
||||
def check_indicators(self) -> None:
|
||||
raise NotImplementedError
|
||||
|
||||
def collect_url_results(self) -> None:
|
||||
"""Collect URL records exposed by this module."""
|
||||
|
||||
def add_url_result(self, url: str, timestamp: Optional[str], source: str) -> None:
|
||||
expanded_url = None
|
||||
if self.indicators:
|
||||
expanded_url = self.indicators.get_expanded_url(url)
|
||||
|
||||
self.url_results.append(
|
||||
{
|
||||
"url": url,
|
||||
"expanded_url": expanded_url,
|
||||
"timestamp": timestamp,
|
||||
"source": source,
|
||||
}
|
||||
)
|
||||
|
||||
def save_to_json(self) -> None:
|
||||
if not self.results_path:
|
||||
return
|
||||
@@ -249,6 +268,15 @@ def run_module(module: MVTModule) -> None:
|
||||
"The %s module produced no detections!", module.__class__.__name__
|
||||
)
|
||||
|
||||
try:
|
||||
module.collect_url_results()
|
||||
except Exception as exc:
|
||||
module.log.exception(
|
||||
"Error when collecting URLs from module %s: %s",
|
||||
module.__class__.__name__,
|
||||
exc,
|
||||
)
|
||||
|
||||
try:
|
||||
module.to_timeline()
|
||||
except NotImplementedError:
|
||||
|
||||
@@ -6,6 +6,7 @@
|
||||
import hashlib
|
||||
import importlib.util
|
||||
import inspect
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
@@ -15,6 +16,7 @@ from typing import Iterable, Optional
|
||||
from .module import MVTModule
|
||||
|
||||
MVT_CUSTOM_MODULES_ENV = "MVT_CUSTOM_MODULES"
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class CustomModuleLoadError(Exception):
|
||||
@@ -122,6 +124,11 @@ def module_supports_command(
|
||||
) -> bool:
|
||||
supported_commands = getattr(module_class, "supported_commands", None)
|
||||
if not supported_commands:
|
||||
return True
|
||||
log.warning(
|
||||
"Custom module %s has no supported_commands and will not be run. "
|
||||
"Declare the platform/command pairs it supports.",
|
||||
module_class.__name__,
|
||||
)
|
||||
return False
|
||||
|
||||
return (platform, command) in {tuple(entry) for entry in supported_commands}
|
||||
@@ -4,7 +4,7 @@
|
||||
# https://license.mvt.re/1.1/
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Dict, List, Union
|
||||
from typing import Any, Dict, List, Optional, TypedDict, Union
|
||||
|
||||
|
||||
# ModuleAtomicResult is a flexible dictionary that can contain any data.
|
||||
@@ -22,6 +22,13 @@ ModuleAtomicResult = Dict[str, Any]
|
||||
ModuleResults = Any
|
||||
|
||||
|
||||
class URLResult(TypedDict):
|
||||
url: str
|
||||
expanded_url: Optional[str]
|
||||
timestamp: Optional[str]
|
||||
source: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class ModuleAtomicTimeline:
|
||||
timestamp: str
|
||||
|
||||
@@ -3,4 +3,4 @@
|
||||
# Use of this software is governed by the MVT License 1.1 that can be found at
|
||||
# https://license.mvt.re/1.1/
|
||||
|
||||
MVT_VERSION = "2026.5.12"
|
||||
MVT_VERSION = "2026.7.29"
|
||||
@@ -42,6 +42,7 @@ from mvt.common.help import (
|
||||
HELP_MSG_CHECK_IOCS,
|
||||
HELP_MSG_STIX2,
|
||||
HELP_MSG_CHECK_IOS_BACKUP,
|
||||
HELP_MSG_CHECK_SYSDIAGNOSE,
|
||||
HELP_MSG_DISABLE_UPDATE_CHECK,
|
||||
HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK,
|
||||
HELP_MSG_COMPLETION,
|
||||
@@ -50,6 +51,7 @@ from mvt.common.module_loader import CustomModuleLoadError, load_custom_modules
|
||||
from mvt.common.password import prompt_password
|
||||
from .cmd_check_backup import CmdIOSCheckBackup
|
||||
from .cmd_check_fs import CmdIOSCheckFS
|
||||
from .cmd_check_sysdiagnose import CmdIOSCheckSysdiagnose
|
||||
from .decrypt import DecryptBackup
|
||||
from .modules.backup import BACKUP_MODULES
|
||||
from .modules.fs import FS_MODULES
|
||||
@@ -398,6 +400,77 @@ def check_fs(
|
||||
cmd.show_support_message()
|
||||
|
||||
|
||||
# ==============================================================================
|
||||
# Command: check-sysdiagnose
|
||||
# ==============================================================================
|
||||
@cli.command(
|
||||
"check-sysdiagnose",
|
||||
context_settings=CONTEXT_SETTINGS,
|
||||
help=HELP_MSG_CHECK_SYSDIAGNOSE,
|
||||
)
|
||||
@click.option(
|
||||
"--iocs",
|
||||
"-i",
|
||||
type=click.Path(exists=True),
|
||||
multiple=True,
|
||||
default=[],
|
||||
help=HELP_MSG_IOC,
|
||||
)
|
||||
@click.option("--output", "-o", type=click.Path(exists=False), help=HELP_MSG_OUTPUT)
|
||||
@click.option("--list-modules", "-l", is_flag=True, help=HELP_MSG_LIST_MODULES)
|
||||
@click.option("--module", "-m", help=HELP_MSG_MODULE)
|
||||
@click.option(
|
||||
"--load-module",
|
||||
type=click.Path(exists=True),
|
||||
multiple=True,
|
||||
default=[],
|
||||
help=HELP_MSG_LOAD_MODULE,
|
||||
)
|
||||
@click.option("--hashes", "-H", is_flag=True, help=HELP_MSG_HASHES)
|
||||
@click.option("--verbose", "-v", is_flag=True, help=HELP_MSG_VERBOSE)
|
||||
@click.argument("SYSDIAGNOSE_PATH", type=click.Path(exists=True))
|
||||
@click.pass_context
|
||||
def check_sysdiagnose(
|
||||
ctx,
|
||||
iocs,
|
||||
output,
|
||||
list_modules,
|
||||
module,
|
||||
load_module,
|
||||
hashes,
|
||||
verbose,
|
||||
sysdiagnose_path,
|
||||
):
|
||||
set_verbose_logging(verbose)
|
||||
custom_modules = _load_custom_modules(load_module)
|
||||
cmd = CmdIOSCheckSysdiagnose(
|
||||
target_path=sysdiagnose_path,
|
||||
results_path=output,
|
||||
ioc_files=iocs,
|
||||
module_name=module,
|
||||
hashes=hashes,
|
||||
disable_version_check=_get_disable_flags(ctx)[0],
|
||||
disable_indicator_check=_get_disable_flags(ctx)[1],
|
||||
custom_modules=custom_modules,
|
||||
)
|
||||
|
||||
if not cmd._available_modules():
|
||||
raise click.ClickException(
|
||||
"No custom modules support mvt-ios check-sysdiagnose. "
|
||||
"Load a module that declares supported_commands = "
|
||||
"((\"ios\", \"check-sysdiagnose\"),)."
|
||||
)
|
||||
|
||||
if list_modules:
|
||||
cmd.list_modules()
|
||||
return
|
||||
|
||||
log.info("Checking iOS sysdiagnose at path: %s", sysdiagnose_path)
|
||||
cmd.run()
|
||||
cmd.show_alerts_brief()
|
||||
cmd.show_support_message()
|
||||
|
||||
|
||||
# ==============================================================================
|
||||
# Command: check-iocs
|
||||
# ==============================================================================
|
||||
|
||||
@@ -0,0 +1,119 @@
|
||||
# Mobile Verification Toolkit (MVT)
|
||||
# Copyright (c) 2021-2023 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 json
|
||||
import logging
|
||||
import os
|
||||
import tarfile
|
||||
from pathlib import Path
|
||||
from typing import Any, Optional
|
||||
|
||||
from mvt.common.command import Command
|
||||
from mvt.common.indicators import Indicators
|
||||
from mvt.common.module import MVTModule
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class CmdIOSCheckSysdiagnose(Command):
|
||||
def __init__(
|
||||
self,
|
||||
target_path: Optional[str] = None,
|
||||
results_path: Optional[str] = None,
|
||||
ioc_files: Optional[list] = None,
|
||||
iocs: Optional[Indicators] = None,
|
||||
module_name: Optional[str] = None,
|
||||
serial: Optional[str] = None,
|
||||
module_options: Optional[dict] = None,
|
||||
hashes: bool = False,
|
||||
sub_command: bool = False,
|
||||
disable_version_check: bool = False,
|
||||
disable_indicator_check: bool = False,
|
||||
custom_modules: Optional[list[type[MVTModule]]] = None,
|
||||
) -> None:
|
||||
super().__init__(
|
||||
target_path=target_path,
|
||||
results_path=results_path,
|
||||
ioc_files=ioc_files,
|
||||
iocs=iocs,
|
||||
module_name=module_name,
|
||||
serial=serial,
|
||||
module_options=module_options,
|
||||
hashes=hashes,
|
||||
sub_command=sub_command,
|
||||
log=log,
|
||||
disable_version_check=disable_version_check,
|
||||
disable_indicator_check=disable_indicator_check,
|
||||
custom_modules=custom_modules,
|
||||
)
|
||||
self.platform = "ios"
|
||||
self.name = "check-sysdiagnose"
|
||||
self.sysdiagnose_format: Optional[str] = None
|
||||
self.sysdiagnose_archive: Optional[tarfile.TarFile] = None
|
||||
self.sysdiagnose_files: list[str] = []
|
||||
self.ips_files: list[dict[str, Any]] = []
|
||||
|
||||
@staticmethod
|
||||
def _parse_bugtype_header(data: bytes) -> Optional[int]:
|
||||
try:
|
||||
header = json.loads(data.split(b"\n", 1)[0].decode("utf-8"))
|
||||
return int(header["bug_type"])
|
||||
except (json.JSONDecodeError, KeyError, UnicodeDecodeError, ValueError):
|
||||
return None
|
||||
|
||||
def _add_ips_file(self, file_path: str, data: bytes) -> None:
|
||||
bug_type = self._parse_bugtype_header(data)
|
||||
if bug_type is not None:
|
||||
self.ips_files.append({"file_path": file_path, "bug_type": bug_type})
|
||||
|
||||
def init(self) -> None:
|
||||
if not self.target_path:
|
||||
raise ValueError("A sysdiagnose path is required")
|
||||
|
||||
if os.path.isdir(self.target_path):
|
||||
self.sysdiagnose_format = "dir"
|
||||
parent_path = Path(self.target_path).absolute().parent
|
||||
for root, _, filenames in os.walk(self.target_path):
|
||||
for filename in filenames:
|
||||
absolute_path = os.path.join(root, filename)
|
||||
file_path = os.path.relpath(absolute_path, parent_path)
|
||||
self.sysdiagnose_files.append(file_path)
|
||||
if filename.endswith(".ips"):
|
||||
with open(absolute_path, "rb") as handle:
|
||||
self._add_ips_file(absolute_path, handle.read())
|
||||
return
|
||||
|
||||
if not os.path.isfile(self.target_path):
|
||||
raise ValueError(f"Sysdiagnose path does not exist: {self.target_path}")
|
||||
|
||||
self.log.info("Parsing sysdiagnose archive. This might take a while...")
|
||||
self.sysdiagnose_format = "tar"
|
||||
self.sysdiagnose_archive = tarfile.open(self.target_path, "r:gz")
|
||||
for member in self.sysdiagnose_archive:
|
||||
self.sysdiagnose_files.append(member.name)
|
||||
if member.isfile() and member.name.endswith(".ips"):
|
||||
archive_handle = self.sysdiagnose_archive.extractfile(member)
|
||||
if archive_handle is not None:
|
||||
with archive_handle:
|
||||
self._add_ips_file(member.name, archive_handle.read())
|
||||
|
||||
def module_init(self, module) -> None:
|
||||
module.ips_files = self.ips_files
|
||||
if self.sysdiagnose_format == "tar":
|
||||
if self.sysdiagnose_archive is None:
|
||||
raise RuntimeError("Sysdiagnose archive has not been initialized")
|
||||
module.from_sysdiagnose_tar(
|
||||
self.sysdiagnose_archive, self.sysdiagnose_files
|
||||
)
|
||||
return
|
||||
if self.sysdiagnose_format == "dir" and self.target_path:
|
||||
module.from_sysdiagnose_folder(self.target_path, self.sysdiagnose_files)
|
||||
return
|
||||
raise RuntimeError("Sysdiagnose input has not been initialized")
|
||||
|
||||
def finish(self) -> None:
|
||||
if self.sysdiagnose_archive is not None:
|
||||
self.sysdiagnose_archive.close()
|
||||
self.sysdiagnose_archive = None
|
||||
@@ -101,6 +101,11 @@ class SMS(IOSExtraction):
|
||||
ioc_match.message, "", result, matched_indicator=ioc_match.ioc
|
||||
)
|
||||
|
||||
def collect_url_results(self) -> None:
|
||||
for message in self.results:
|
||||
for url in message.get("links", []):
|
||||
self.add_url_result(url, message.get("isodate"), "sms")
|
||||
|
||||
def run(self) -> None:
|
||||
self._find_ios_database(backup_ids=SMS_BACKUP_IDS, root_paths=SMS_ROOT_PATHS)
|
||||
self.log.info("Found SMS database at path: %s", self.file_path)
|
||||
|
||||
@@ -70,6 +70,11 @@ class Whatsapp(IOSExtraction):
|
||||
ioc_match.message, "", result, matched_indicator=ioc_match.ioc
|
||||
)
|
||||
|
||||
def collect_url_results(self) -> None:
|
||||
for message in self.results:
|
||||
for url in message.get("links", []):
|
||||
self.add_url_result(url, message.get("isodate"), "whatsapp")
|
||||
|
||||
def run(self) -> None:
|
||||
self._find_ios_database(
|
||||
backup_ids=WHATSAPP_BACKUP_IDS, root_paths=WHATSAPP_ROOT_PATHS
|
||||
|
||||
@@ -0,0 +1,6 @@
|
||||
# Mobile Verification Toolkit (MVT)
|
||||
# Copyright (c) 2021-2023 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 .base import SysdiagnoseExtraction
|
||||
@@ -0,0 +1,105 @@
|
||||
# Mobile Verification Toolkit (MVT)
|
||||
# Copyright (c) 2021-2023 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 fnmatch
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import tarfile
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from mvt.common.module import MVTModule, ModuleResults
|
||||
|
||||
|
||||
class SysdiagnoseExtraction(MVTModule):
|
||||
"""Base class for custom modules that analyze an iOS sysdiagnose."""
|
||||
|
||||
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,
|
||||
)
|
||||
self.parent_path: Optional[str] = None
|
||||
self.files: list[str] = []
|
||||
self.tar: Optional[tarfile.TarFile] = None
|
||||
self.tar_files: list[str] = []
|
||||
self.ips_files: list[dict[str, object]] = []
|
||||
|
||||
def from_sysdiagnose_folder(
|
||||
self, target_path: str, sysdiagnose_files: list[str]
|
||||
) -> None:
|
||||
self.parent_path = Path(target_path).absolute().parent.as_posix()
|
||||
self.files = sysdiagnose_files
|
||||
|
||||
def from_sysdiagnose_tar(
|
||||
self, sysdiagnose_archive: tarfile.TarFile, sysdiagnose_files: list[str]
|
||||
) -> None:
|
||||
self.tar = sysdiagnose_archive
|
||||
self.tar_files = sysdiagnose_files
|
||||
|
||||
def _extract_timezone(self):
|
||||
"""Determine the sysdiagnose timezone from its diagnostic log."""
|
||||
file_paths = self._get_files_by_pattern("*/sysdiagnose.log")
|
||||
if not file_paths:
|
||||
self.log.info(
|
||||
"Unable to determine the timezone in which the sysdiagnose was "
|
||||
"generated. Assuming UTC for logs without a timezone."
|
||||
)
|
||||
return timezone.utc
|
||||
|
||||
content = self._get_file_content(file_paths[0]).decode(
|
||||
"utf-8", errors="replace"
|
||||
)
|
||||
filenames = re.findall(r"sysdiagnose_\S+?\.tar\.gz", content)
|
||||
if not filenames:
|
||||
self.log.info(
|
||||
"Unable to determine the timezone in which the sysdiagnose was "
|
||||
"generated. Assuming UTC for logs without a timezone."
|
||||
)
|
||||
return timezone.utc
|
||||
|
||||
timestamp = "_".join(
|
||||
filenames[0].removesuffix(".tar.gz").split("_")[1:3]
|
||||
)
|
||||
sysdiagnose_timezone = datetime.strptime(
|
||||
timestamp, "%Y.%m.%d_%H-%M-%S%z"
|
||||
).tzinfo
|
||||
self.log.info(
|
||||
"Based on the sysdiagnose filename, assuming timezone %s for logs "
|
||||
"without a timezone.",
|
||||
sysdiagnose_timezone,
|
||||
)
|
||||
return sysdiagnose_timezone
|
||||
|
||||
def _get_files_by_pattern(self, pattern: str) -> list[str]:
|
||||
file_names = self.tar_files if self.tar else self.files
|
||||
return fnmatch.filter(file_names, pattern)
|
||||
|
||||
def _get_file_content(self, file_path: str) -> bytes:
|
||||
if self.tar:
|
||||
handle = self.tar.extractfile(self.tar.getmember(file_path))
|
||||
else:
|
||||
if self.parent_path is None:
|
||||
raise RuntimeError("Sysdiagnose folder has not been initialized")
|
||||
handle = open(os.path.join(self.parent_path, file_path), "rb")
|
||||
|
||||
if handle is None:
|
||||
return b""
|
||||
with handle:
|
||||
return handle.read()
|
||||
@@ -4,6 +4,8 @@
|
||||
# https://license.mvt.re/1.1/
|
||||
|
||||
from mvt.android.artifacts.dumpsys_adb import DumpsysADBArtifact
|
||||
from mvt.android.modules.bugreport.dumpsys_adb_state import DumpsysADBState
|
||||
from mvt.common.alerts import AlertLevel
|
||||
|
||||
from ..utils import get_artifact
|
||||
|
||||
@@ -114,3 +116,119 @@ class TestDumpsysADBArtifact:
|
||||
assert key_store_entry["user"] == "user@laptop"
|
||||
assert key_store_entry["fingerprint"] == expected_fingerprint
|
||||
assert key_store_entry["last_connected"] == "1628501829898"
|
||||
|
||||
|
||||
class TestDumpsysADBStateAlerts:
|
||||
def test_no_androidqf_context_preserves_existing_behavior(self):
|
||||
module = DumpsysADBState(
|
||||
results=[
|
||||
{
|
||||
"user_keys": [
|
||||
{
|
||||
"key": b"QUJDRA==",
|
||||
"user": "host@example",
|
||||
"fingerprint": "fingerprint",
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
)
|
||||
|
||||
module.check_indicators()
|
||||
|
||||
assert module.alertstore.alerts == []
|
||||
|
||||
def test_androidqf_trusted_keys_create_expected_alerts(self):
|
||||
module = DumpsysADBState(
|
||||
module_options={
|
||||
"androidqf_acquisition": {
|
||||
"started": "2025-06-20T18:00:00Z",
|
||||
"adb_host_public_key": "QUJDRA== acquisition@host",
|
||||
}
|
||||
},
|
||||
results=[
|
||||
{
|
||||
"user_keys": [
|
||||
{
|
||||
"key": b"QUJDRA==",
|
||||
"user": "acquisition@host",
|
||||
"fingerprint": "acquisition-fingerprint",
|
||||
},
|
||||
{
|
||||
"key": b"RUZHSA==",
|
||||
"user": "other@host",
|
||||
"fingerprint": "other-fingerprint",
|
||||
},
|
||||
{
|
||||
"key": b"not-base64",
|
||||
"user": "invalid@host",
|
||||
"fingerprint": "",
|
||||
},
|
||||
],
|
||||
"keystore": [
|
||||
{
|
||||
"key": b"QUJDRA==",
|
||||
"user": "acquisition@host",
|
||||
"fingerprint": "acquisition-fingerprint",
|
||||
"last_connected": "1750266000000",
|
||||
}
|
||||
],
|
||||
}
|
||||
],
|
||||
)
|
||||
|
||||
module.check_indicators()
|
||||
|
||||
assert [alert.level for alert in module.alertstore.alerts] == [
|
||||
AlertLevel.INFORMATIONAL,
|
||||
AlertLevel.LOW,
|
||||
AlertLevel.LOW,
|
||||
]
|
||||
informational, different, invalid = module.alertstore.alerts
|
||||
assert "at least one day before" in informational.message
|
||||
assert informational.event_time == "2025-06-18 17:00:00.000000"
|
||||
assert "different from the AndroidQF acquisition host" in different.message
|
||||
assert "invalid trusted ADB host key" in invalid.message
|
||||
|
||||
def test_missing_androidqf_host_key_creates_low_alert(self):
|
||||
trusted_key = {
|
||||
"key": b"QUJDRA==",
|
||||
"user": "host@example",
|
||||
"fingerprint": "fingerprint",
|
||||
}
|
||||
module = DumpsysADBState(
|
||||
module_options={"androidqf_acquisition": {}},
|
||||
results=[{"user_keys": [trusted_key]}],
|
||||
)
|
||||
|
||||
module.check_indicators()
|
||||
|
||||
assert len(module.alertstore.alerts) == 1
|
||||
assert module.alertstore.alerts[0].level == AlertLevel.LOW
|
||||
assert "does not include its host key" in module.alertstore.alerts[0].message
|
||||
|
||||
def test_recent_acquisition_host_key_does_not_create_alert(self):
|
||||
module = DumpsysADBState(
|
||||
module_options={
|
||||
"androidqf_acquisition": {
|
||||
"started": "2025-06-20T18:00:00Z",
|
||||
"adb_host_public_key": "QUJDRA== acquisition@host",
|
||||
}
|
||||
},
|
||||
results=[
|
||||
{
|
||||
"keystore": [
|
||||
{
|
||||
"key": b"QUJDRA==",
|
||||
"user": "acquisition@host",
|
||||
"fingerprint": "fingerprint",
|
||||
"last_connected": "1750438800000",
|
||||
}
|
||||
]
|
||||
}
|
||||
],
|
||||
)
|
||||
|
||||
module.check_indicators()
|
||||
|
||||
assert module.alertstore.alerts == []
|
||||
@@ -42,6 +42,15 @@ class IndependentModule(RecordingModule):
|
||||
pass
|
||||
|
||||
|
||||
class URLRecordingModule(RecordingModule):
|
||||
def collect_url_results(self):
|
||||
self.add_url_result(
|
||||
"https://example.org/message",
|
||||
"2026-07-29 12:00:00.000000",
|
||||
"test-chat",
|
||||
)
|
||||
|
||||
|
||||
class CustomIOSBackupModule(RecordingModule):
|
||||
supported_commands = (("ios", "check-backup"),)
|
||||
|
||||
@@ -50,6 +59,10 @@ class CustomIOSFSModule(RecordingModule):
|
||||
supported_commands = (("ios", "check-fs"),)
|
||||
|
||||
|
||||
class UnscopedCustomModule(RecordingModule):
|
||||
pass
|
||||
|
||||
|
||||
class CustomDependsOnBuiltin(RecordingModule):
|
||||
supported_commands = (("ios", "check-backup"),)
|
||||
dependencies = (FirstModule,)
|
||||
@@ -83,6 +96,21 @@ class TestCommand:
|
||||
alerts = json.loads((tmp_path / "alerts.json").read_text())
|
||||
assert alerts[0]["event"]["payload"] == "\\xa8\\xa9"
|
||||
|
||||
def test_stores_collected_urls(self, tmp_path):
|
||||
cmd = RecordingCommand(results_path=str(tmp_path))
|
||||
cmd.modules = [URLRecordingModule]
|
||||
|
||||
cmd.run()
|
||||
|
||||
assert json.loads((tmp_path / "urls.json").read_text()) == [
|
||||
{
|
||||
"url": "https://example.org/message",
|
||||
"expanded_url": None,
|
||||
"timestamp": "2026-07-29 12:00:00.000000",
|
||||
"source": "test-chat",
|
||||
}
|
||||
]
|
||||
|
||||
def test_modules_run_in_stable_topological_order(self):
|
||||
cmd = RecordingCommand()
|
||||
cmd.modules = [ThirdModule, IndependentModule, SecondModule, FirstModule]
|
||||
@@ -151,7 +179,11 @@ class TestCommand:
|
||||
cmd.platform = "ios"
|
||||
cmd.name = "check-backup"
|
||||
cmd.modules = [FirstModule]
|
||||
cmd.custom_modules = [CustomIOSBackupModule, CustomIOSFSModule]
|
||||
cmd.custom_modules = [
|
||||
CustomIOSBackupModule,
|
||||
CustomIOSFSModule,
|
||||
UnscopedCustomModule,
|
||||
]
|
||||
|
||||
assert [module.__name__ for module in cmd._ordered_modules()] == [
|
||||
"FirstModule",
|
||||
|
||||
@@ -197,6 +197,15 @@ class TestIndicators:
|
||||
assert matches[0] is None
|
||||
assert matches[1]
|
||||
assert matches[1].ioc.value == "example.org"
|
||||
assert (
|
||||
ind.get_expanded_url("https://tinyurl.com/nested")
|
||||
== "https://www.example.org/landing"
|
||||
)
|
||||
assert (
|
||||
ind.get_expanded_url("https://t.co/nested")
|
||||
== "https://www.example.org/landing"
|
||||
)
|
||||
assert ind.get_expanded_url("https://bit.ly/failure") is None
|
||||
assert {call.args[0] for call in head.call_args_list} == {
|
||||
"https://bit.ly/failure",
|
||||
"https://tinyurl.com/nested",
|
||||
|
||||
@@ -125,12 +125,13 @@ def test_load_custom_modules_loads_env_folder_first(tmp_path, monkeypatch):
|
||||
assert [module.__name__ for module in modules] == ["EnvModule", "CliModule"]
|
||||
|
||||
|
||||
def test_module_supports_command_defaults_to_all_commands(tmp_path):
|
||||
def test_module_supports_command_requires_explicit_declaration(tmp_path, caplog):
|
||||
module_path = _write_module(tmp_path / "custom.py", "DefaultModule")
|
||||
module = load_custom_modules_from_path(str(module_path))[0]
|
||||
|
||||
assert module_supports_command(module, "ios", "check-backup")
|
||||
assert module_supports_command(module, "android", "check-bugreport")
|
||||
assert not module_supports_command(module, "ios", "check-backup")
|
||||
assert not module_supports_command(module, "android", "check-bugreport")
|
||||
assert "DefaultModule has no supported_commands" in caplog.text
|
||||
|
||||
|
||||
def test_module_supports_command_honors_supported_commands(tmp_path):
|
||||
|
||||
@@ -18,6 +18,14 @@ class TestSMSModule:
|
||||
run_module(m)
|
||||
assert len(m.results) == 1
|
||||
assert len(m.timeline) == 2
|
||||
assert m.url_results == [
|
||||
{
|
||||
"url": "https://badbadbad.example.org/",
|
||||
"expanded_url": None,
|
||||
"timestamp": "2019-08-29 23:13:30.000000",
|
||||
"source": "sms",
|
||||
}
|
||||
]
|
||||
assert len(m.alertstore.alerts) == 0
|
||||
|
||||
def test_detection(self, indicator_file):
|
||||
|
||||
@@ -0,0 +1,35 @@
|
||||
# Mobile Verification Toolkit (MVT)
|
||||
# Copyright (c) 2021-2026 The MVT Authors.
|
||||
# Use of this software is governed by the MVT License 1.1 that can be found at
|
||||
# https://license.mvt.re/1.1/
|
||||
|
||||
import logging
|
||||
|
||||
from mvt.common.indicators import Indicators
|
||||
from mvt.ios.modules.mixed.whatsapp import Whatsapp
|
||||
|
||||
|
||||
def test_collect_url_results_includes_expansion():
|
||||
module = Whatsapp(
|
||||
results=[
|
||||
{
|
||||
"links": ["https://bit.ly/message"],
|
||||
"isodate": "2026-07-29 12:00:00.000000",
|
||||
}
|
||||
]
|
||||
)
|
||||
module.indicators = Indicators(log=logging.getLogger())
|
||||
module.indicators.resolved_urls["https://bit.ly/message"] = (
|
||||
"https://example.org/landing"
|
||||
)
|
||||
|
||||
module.collect_url_results()
|
||||
|
||||
assert module.url_results == [
|
||||
{
|
||||
"url": "https://bit.ly/message",
|
||||
"expanded_url": "https://example.org/landing",
|
||||
"timestamp": "2026-07-29 12:00:00.000000",
|
||||
"source": "whatsapp",
|
||||
}
|
||||
]
|
||||
@@ -3,6 +3,7 @@
|
||||
# Use of this software is governed by the MVT License 1.1 that can be found at
|
||||
# https://license.mvt.re/1.1/
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
@@ -32,6 +33,67 @@ class TestCheckAndroidqfCommand:
|
||||
result = runner.invoke(check_androidqf, [path])
|
||||
assert result.exit_code == 0
|
||||
|
||||
def test_check_stores_nested_sms_urls(self, tmp_path):
|
||||
runner = CliRunner()
|
||||
path = os.path.join(get_artifact_folder(), "androidqf")
|
||||
|
||||
result = runner.invoke(check_androidqf, ["--output", str(tmp_path), path])
|
||||
|
||||
assert result.exit_code == 0
|
||||
urls = json.loads((tmp_path / "urls.json").read_text())
|
||||
assert {entry["url"] for entry in urls} == {
|
||||
"http://google.com",
|
||||
"https://google.com/",
|
||||
}
|
||||
assert all(
|
||||
set(entry) == {"url", "expanded_url", "timestamp", "source"}
|
||||
for entry in urls
|
||||
)
|
||||
assert all(entry["source"] == "sms" for entry in urls)
|
||||
|
||||
def test_acquisition_context_is_passed_to_bugreport(self, tmp_path, mocker):
|
||||
data_path = tmp_path / "androidqf"
|
||||
data_path.mkdir()
|
||||
(data_path / "acquisition.json").write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"started": "2025-06-20T18:00:00Z",
|
||||
"adb_host_public_key": "QUJDRA== acquisition@host",
|
||||
}
|
||||
)
|
||||
)
|
||||
with zipfile.ZipFile(data_path / "bugreport.zip", "w"):
|
||||
pass
|
||||
|
||||
nested_command = mocker.patch(
|
||||
"mvt.android.cmd_check_androidqf.CmdAndroidCheckBugreport"
|
||||
)
|
||||
nested_command.return_value.timeline = []
|
||||
nested_command.return_value.alertstore.alerts = []
|
||||
command = CmdAndroidCheckAndroidQF(target_path=str(data_path))
|
||||
command.init()
|
||||
|
||||
assert command.run_bugreport_cmd() is True
|
||||
|
||||
assert nested_command.call_args.kwargs["module_options"][
|
||||
"androidqf_acquisition"
|
||||
] == {
|
||||
"started": "2025-06-20T18:00:00Z",
|
||||
"adb_host_public_key": "QUJDRA== acquisition@host",
|
||||
}
|
||||
|
||||
def test_acquisition_context_falls_back_to_public_key_file(self, tmp_path):
|
||||
data_path = tmp_path / "androidqf"
|
||||
data_path.mkdir()
|
||||
(data_path / "adb_host_key.pub").write_text("QUJDRA== acquisition@host\n")
|
||||
command = CmdAndroidCheckAndroidQF(target_path=str(data_path))
|
||||
|
||||
command.init()
|
||||
|
||||
assert command.module_options["androidqf_acquisition"] == {
|
||||
"adb_host_public_key": "QUJDRA== acquisition@host\n"
|
||||
}
|
||||
|
||||
def test_check_encrypted_backup_prompt_valid(self, mocker):
|
||||
"""Prompt for password on CLI"""
|
||||
prompt_mock = mocker.patch(
|
||||
|
||||
@@ -0,0 +1,57 @@
|
||||
from click.testing import CliRunner
|
||||
|
||||
from mvt.ios.cli import check_sysdiagnose
|
||||
|
||||
|
||||
CUSTOM_MODULE = """
|
||||
from mvt.ios.modules.sysdiagnose import SysdiagnoseExtraction
|
||||
|
||||
|
||||
class CustomSysdiagnoseModule(SysdiagnoseExtraction):
|
||||
supported_commands = (("ios", "check-sysdiagnose"),)
|
||||
slug = "custom_sysdiagnose_module"
|
||||
|
||||
def run(self):
|
||||
file_path = self._get_files_by_pattern("*/artifact.txt")[0]
|
||||
self.results = [{"content": self._get_file_content(file_path).decode("utf-8")}]
|
||||
|
||||
def check_indicators(self):
|
||||
pass
|
||||
|
||||
def serialize(self, result):
|
||||
return None
|
||||
"""
|
||||
|
||||
|
||||
def _create_sysdiagnose_folder(tmp_path):
|
||||
folder = tmp_path / "sysdiagnose"
|
||||
folder.mkdir()
|
||||
(folder / "artifact.txt").write_text("artifact", encoding="utf-8")
|
||||
return folder
|
||||
|
||||
|
||||
def test_check_sysdiagnose_runs_explicitly_scoped_custom_module(tmp_path):
|
||||
module_path = tmp_path / "custom_sysdiagnose.py"
|
||||
module_path.write_text(CUSTOM_MODULE, encoding="utf-8")
|
||||
output_path = tmp_path / "output"
|
||||
|
||||
result = CliRunner().invoke(
|
||||
check_sysdiagnose,
|
||||
[
|
||||
"--load-module",
|
||||
str(module_path),
|
||||
"--output",
|
||||
str(output_path),
|
||||
str(_create_sysdiagnose_folder(tmp_path)),
|
||||
],
|
||||
)
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert (output_path / "custom_sysdiagnose_module.json").exists()
|
||||
|
||||
|
||||
def test_check_sysdiagnose_requires_an_explicitly_scoped_module(tmp_path):
|
||||
result = CliRunner().invoke(check_sysdiagnose, [str(_create_sysdiagnose_folder(tmp_path))])
|
||||
|
||||
assert result.exit_code != 0
|
||||
assert "No custom modules support mvt-ios check-sysdiagnose" in result.output
|
||||
@@ -0,0 +1,75 @@
|
||||
import tarfile
|
||||
from datetime import timedelta
|
||||
|
||||
from mvt.ios.cmd_check_sysdiagnose import CmdIOSCheckSysdiagnose
|
||||
from mvt.ios.modules.sysdiagnose import SysdiagnoseExtraction
|
||||
|
||||
|
||||
class SysdiagnoseTestModule(SysdiagnoseExtraction):
|
||||
supported_commands = (("ios", "check-sysdiagnose"),)
|
||||
|
||||
def run(self):
|
||||
file_path = self._get_files_by_pattern("*/artifact.txt")[0]
|
||||
self.results = [
|
||||
{
|
||||
"content": self._get_file_content(file_path).decode("utf-8"),
|
||||
"timezone_offset": self._extract_timezone().utcoffset(None).seconds,
|
||||
}
|
||||
]
|
||||
|
||||
def check_indicators(self):
|
||||
pass
|
||||
|
||||
def serialize(self, result):
|
||||
return None
|
||||
|
||||
|
||||
def _create_sysdiagnose_folder(tmp_path):
|
||||
folder = tmp_path / "sysdiagnose"
|
||||
folder.mkdir()
|
||||
(folder / "artifact.txt").write_text("artifact", encoding="utf-8")
|
||||
(folder / "sysdiagnose.log").write_text(
|
||||
"sysdiagnose_2024.01.02_03-04-05+0200.tar.gz", encoding="utf-8"
|
||||
)
|
||||
(folder / "report.ips").write_text('{"bug_type": 210}\nbody', encoding="utf-8")
|
||||
return folder
|
||||
|
||||
|
||||
def _create_sysdiagnose_archive(tmp_path, folder):
|
||||
archive_path = tmp_path / "sysdiagnose.tar.gz"
|
||||
with tarfile.open(archive_path, "w:gz") as archive:
|
||||
for path in folder.iterdir():
|
||||
archive.add(path, arcname=f"sysdiagnose/{path.name}")
|
||||
return archive_path
|
||||
|
||||
|
||||
def _run_command(path):
|
||||
command = CmdIOSCheckSysdiagnose(
|
||||
target_path=str(path), custom_modules=[SysdiagnoseTestModule]
|
||||
)
|
||||
command.run()
|
||||
return command
|
||||
|
||||
|
||||
def test_check_sysdiagnose_from_folder(tmp_path):
|
||||
command = _run_command(_create_sysdiagnose_folder(tmp_path))
|
||||
|
||||
assert command.executed[0].results == [
|
||||
{"content": "artifact", "timezone_offset": timedelta(hours=2).seconds}
|
||||
]
|
||||
assert command.executed[0].ips_files == [
|
||||
{"file_path": str(tmp_path / "sysdiagnose" / "report.ips"), "bug_type": 210}
|
||||
]
|
||||
|
||||
|
||||
def test_check_sysdiagnose_from_archive_closes_archive(tmp_path):
|
||||
folder = _create_sysdiagnose_folder(tmp_path)
|
||||
command = _run_command(_create_sysdiagnose_archive(tmp_path, folder))
|
||||
|
||||
assert command.executed[0].results == [
|
||||
{"content": "artifact", "timezone_offset": timedelta(hours=2).seconds}
|
||||
]
|
||||
assert command.executed[0].ips_files == [
|
||||
{"file_path": "sysdiagnose/report.ips", "bug_type": 210}
|
||||
]
|
||||
assert command.sysdiagnose_archive is None
|
||||
Reference in new issue
Block a user