Compare commits

..
Author SHA1 Message Date
besendorf c57179a62c Merge branch 'main' into add-shell-completion-command 2026-07-01 12:34:36 +02:00
Janik Besendorf f95f7b1283 Add shell completion command 2026-06-19 16:51:01 +02:00
15 changed files with 10 additions and 407 deletions

No files matched your search

-16
View File
@@ -38,22 +38,6 @@ By separating artifact collection from forensic analysis, this approach ensures
For more information, refer to the [AndroidQF project documentation](https://github.com/mvt-project/androidqf).
### VirusTotal package lookups
AndroidQF records APK file hashes in `packages.json`. MVT can optionally look up non-system APK hashes on VirusTotal while checking an AndroidQF acquisition:
```bash
MVT_VT_API_KEY=<key> mvt-android check-androidqf --virustotal /path/to/androidqf-output
```
The `--virustotal` option is disabled by default because it sends APK hashes to VirusTotal and requires network access. It uses the `VT_API_KEY` MVT configuration value, which can also be provided through the `MVT_VT_API_KEY` environment variable.
To avoid exhausting free VirusTotal API quotas, MVT waits 16 seconds between package hash requests by default. Use `--delay` to change the delay, or `--delay 0` to disable throttling:
```bash
mvt-android check-androidqf --virustotal --delay 30 /path/to/androidqf-output
```
## Android Intrusion Logs
On devices where the user has opted into Android's [**Advanced Protection Mode**](https://support.google.com/android/answer/16339980) and turned on the optional Intrusion Logging featrue, Android can create and archive structured *Intrusion Logs* in an encrypted format. These logs record DNS queries, outbound network connections, process starts, ADB activity and other security-relevant events, and are a high-fidelity complement to the rest of an AndroidQF acquisition. The logs are generated on-device and encrypted before being stored in the Google account associated with the device. The encryption key is protected by the user device PIN. The intrusion log data is not accessible to Google.
-10
View File
@@ -23,7 +23,6 @@ from mvt.common.help import (
HELP_MSG_CHECK_BUGREPORT,
HELP_MSG_CHECK_IOCS,
HELP_MSG_CHECK_INTRUSION_LOGS,
HELP_MSG_DELAY_CHECKS,
HELP_MSG_COMPLETION,
HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK,
HELP_MSG_DISABLE_UPDATE_CHECK,
@@ -37,7 +36,6 @@ from mvt.common.help import (
HELP_MSG_STIX2,
HELP_MSG_VERBOSE,
HELP_MSG_VERSION,
HELP_MSG_VIRUS_TOTAL,
)
from mvt.common.logo import logo
from mvt.common.module_loader import CustomModuleLoadError, load_custom_modules
@@ -312,10 +310,6 @@ def check_backup(
help=HELP_MSG_LOAD_MODULE,
)
@click.option("--hashes", "-H", is_flag=True, help=HELP_MSG_HASHES)
@click.option("--virustotal", "-V", is_flag=True, help=HELP_MSG_VIRUS_TOTAL)
@click.option(
"--delay", "-d", type=click.IntRange(min=0), default=16, help=HELP_MSG_DELAY_CHECKS
)
@click.option("--non-interactive", "-n", is_flag=True, help=HELP_MSG_NONINTERACTIVE)
@click.option("--backup-password", "-p", help=HELP_MSG_ANDROID_BACKUP_PASSWORD)
@click.option("--verbose", "-v", is_flag=True, help=HELP_MSG_VERBOSE)
@@ -329,8 +323,6 @@ def check_androidqf(
module,
load_module,
hashes,
virustotal,
delay,
non_interactive,
backup_password,
verbose,
@@ -348,8 +340,6 @@ def check_androidqf(
module_options={
"interactive": not non_interactive,
"backup_password": cli_load_android_backup_password(log, backup_password),
"virustotal": virustotal,
"virustotal_delay": delay,
},
disable_version_check=_get_disable_flags(ctx)[0],
disable_indicator_check=_get_disable_flags(ctx)[1],
@@ -5,11 +5,8 @@
import json
import logging
import time
from typing import Optional
from rich.progress import track
from mvt.android.utils import (
BROWSER_INSTALLERS,
PLAY_STORE_INSTALLERS,
@@ -18,8 +15,7 @@ from mvt.android.utils import (
SYSTEM_UPDATE_PACKAGES,
THIRD_PARTY_STORE_INSTALLERS,
)
from mvt.common.module_types import ModuleAtomicResult, ModuleResults
from mvt.common.virustotal import VTNoKey, VTQuotaExceeded, virustotal_lookup
from mvt.common.module_types import ModuleResults
from .base import AndroidQFModule
@@ -129,65 +125,6 @@ class AQFPackages(AndroidQFModule):
)
break
if self.module_options.get("virustotal", False):
self.check_virustotal(
delay=self.module_options.get("virustotal_delay", 0)
)
def check_virustotal(self, delay: int = 0) -> None:
files_by_hash: dict[
str, list[tuple[ModuleAtomicResult, ModuleAtomicResult]]
] = {}
for package in self.results:
if package.get("system", False):
continue
for package_file in package.get("files", []):
file_hash = package_file.get("sha256")
if not file_hash:
continue
files_by_hash.setdefault(file_hash, []).append((package, package_file))
total_hashes = len(files_by_hash)
if total_hashes == 0:
return
progress_desc = f"Looking up {total_hashes} package files on VirusTotal..."
for index, file_hash in enumerate(
track(files_by_hash, description=progress_desc)
):
try:
results = virustotal_lookup(file_hash)
except VTNoKey as exc:
self.log.warning("%s", exc)
return
except VTQuotaExceeded as exc:
self.log.warning("Unable to continue VirusTotal lookups: %s", exc)
break
if index < total_hashes - 1 and delay > 0:
time.sleep(delay)
if not results:
continue
attributes = results.get("attributes", {})
stats = attributes.get("last_analysis_stats", {})
positives = stats.get("malicious", 0)
total = len(attributes.get("last_analysis_results", {}))
detection = f"{positives}/{total}"
for package, package_file in files_by_hash[file_hash]:
package_file["virustotal"] = detection
if positives > 0:
self.alertstore.high(
f'VirusTotal flagged package "{package["name"]}" file '
f'"{package_file["path"]}" with {detection} detections',
"",
package,
)
def run(self) -> None:
packages = self._get_files_by_pattern("*/packages.json")
if not packages:
@@ -270,11 +270,6 @@ SECURITY_EVENT_METADATA_KEYS = {
"timestamp",
}
# Known platform failures that remain in the timeline but do not need a warning.
KNOWN_BENIGN_KEY_GENERATION_FAILURES = {
("PinStorage_crossReboot_key", 1001),
}
class SecurityEvent(IntrusionLogsModule):
"""This module analyzes security events from intrusion logs."""
@@ -405,17 +400,10 @@ class SecurityEvent(IntrusionLogsModule):
independent of any loaded indicators."""
# Flag failed cryptographic operations as potentially suspicious
if "key_generated" in result:
key_info = result["key_generated"]
key_id = key_info.get("key_id", "unknown")
uid = key_info.get("uid")
is_known_benign_failure = (
key_id,
uid,
) in KNOWN_BENIGN_KEY_GENERATION_FAILURES
if not key_info.get("success", True) and not is_known_benign_failure:
if not result["key_generated"].get("success", True):
self.log.warning(
"Failed key generation detected for key_id: %s",
key_id,
result["key_generated"].get("key_id", "unknown"),
)
# Flag certificate validation failures
-2
View File
@@ -53,5 +53,3 @@ HELP_MSG_CHECK_BUGREPORT = "Check an Android Bug Report"
HELP_MSG_CHECK_ANDROID_BACKUP = "Check an Android Backup"
HELP_MSG_CHECK_ANDROIDQF = "Check data collected with AndroidQF"
HELP_MSG_CHECK_INTRUSION_LOGS = "Check Android Intrusion Logging files"
HELP_MSG_VIRUS_TOTAL = "Check package APK hashes on VirusTotal"
HELP_MSG_DELAY_CHECKS = "Delay in seconds between VirusTotal requests"
-53
View File
@@ -1,53 +0,0 @@
# 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 logging
from typing import Any, Optional
import requests
from .config import settings
log = logging.getLogger(__name__)
class VTNoKey(Exception):
pass
class VTQuotaExceeded(Exception):
pass
def virustotal_lookup(file_hash: str) -> Optional[dict[str, Any]]:
if not settings.VT_API_KEY:
raise VTNoKey(
"No VirusTotal API key provided: to use VirusTotal lookups please set "
"MVT_VT_API_KEY or VT_API_KEY in the MVT configuration file"
)
headers = {
"User-Agent": "VirusTotal",
"Content-Type": "application/json",
"x-apikey": settings.VT_API_KEY,
}
res = requests.get(
f"https://www.virustotal.com/api/v3/files/{file_hash}",
headers=headers,
timeout=settings.NETWORK_TIMEOUT,
)
if res.status_code == 200:
report = res.json()
return report["data"]
if res.status_code == 404:
log.info("Could not find results for file with hash %s", file_hash)
elif res.status_code == 429:
raise VTQuotaExceeded("You have exceeded the quota for your VirusTotal API key")
else:
raise RuntimeError(f"Unexpected response from VirusTotal: {res.status_code}")
return None
+1 -4
View File
@@ -323,10 +323,7 @@ def check_backup(
cmd.list_modules()
return
if not cmd.resolve_backup_path():
ctx.exit(1)
log.info("Checking iTunes backup located at: %s", cmd.target_path)
log.info("Checking iTunes backup located at: %s", backup_path)
cmd.run()
cmd.show_alerts_brief()
-48
View File
@@ -4,7 +4,6 @@
# https://license.mvt.re/1.1/
import logging
import os
from typing import Optional
from mvt.common.command import Command
@@ -17,12 +16,6 @@ from .modules.mixed import MIXED_MODULES
log = logging.getLogger(__name__)
def is_ios_backup_folder(path: str) -> bool:
return os.path.isfile(os.path.join(path, "Manifest.db")) and os.path.isfile(
os.path.join(path, "Info.plist")
)
class CmdIOSCheckBackup(Command):
def __init__(
self,
@@ -59,46 +52,5 @@ class CmdIOSCheckBackup(Command):
self.name = "check-backup"
self.modules = BACKUP_MODULES + MIXED_MODULES
def resolve_backup_path(self) -> bool:
target_path = getattr(self, "target_path", None)
if not isinstance(target_path, str) or not target_path:
return False
if is_ios_backup_folder(target_path):
return True
if not os.path.isdir(target_path):
self.log.critical(
"%s does not appear to be an iTunes backup folder. "
"Expected Manifest.db and Info.plist.",
target_path,
)
return False
candidates = []
for entry_name in sorted(os.listdir(target_path)):
entry_path = os.path.join(target_path, entry_name)
if os.path.isdir(entry_path) and is_ios_backup_folder(entry_path):
candidates.append(entry_path)
if len(candidates) == 1:
self.log.info("Found iTunes backup in subfolder: %s", candidates[0])
self.target_path = candidates[0]
return True
if candidates:
self.log.critical(
"Found multiple iTunes backups in %s. Please specify one backup folder.",
target_path,
)
return False
self.log.critical(
"%s does not appear to be an iTunes backup folder. "
"Expected Manifest.db and Info.plist.",
target_path,
)
return False
def module_init(self, module):
module.is_backup = True
+6 -15
View File
@@ -6,7 +6,6 @@
import logging
from typing import Optional
from mvt.common.module import DatabaseNotFoundError
from mvt.common.module_types import (
ModuleAtomicResult,
ModuleResults,
@@ -18,7 +17,6 @@ from ..base import IOSExtraction
SHUTDOWN_LOG_PATH = [
"private/var/db/diagnostics/shutdown.log",
"private/var/db/diagnostics/shutdown.*.log",
]
@@ -134,17 +132,10 @@ class ShutdownLog(IOSExtraction):
self.results = sorted(self.results, key=lambda entry: entry["isodate"])
def run(self) -> None:
if self.file_path:
shutdown_log_paths = [self.file_path]
else:
shutdown_log_paths = sorted(
self._get_fs_files_from_patterns(SHUTDOWN_LOG_PATH)
)
self._find_ios_database(root_paths=SHUTDOWN_LOG_PATH)
self.log.info("Found shutdown log at path: %s", self.file_path)
if not shutdown_log_paths:
raise DatabaseNotFoundError("unable to find any shutdown log files")
for shutdown_log_path in shutdown_log_paths:
self.log.info("Found shutdown log at path: %s", shutdown_log_path)
with open(shutdown_log_path, "r", encoding="utf-8") as handle:
self.process_shutdownlog(handle.read())
if not self.file_path:
return
with open(self.file_path, "r", encoding="utf-8") as handle:
self.process_shutdownlog(handle.read())
-46
View File
@@ -215,52 +215,6 @@ def _run_security_heuristics(results):
return module.alertstore.alerts
@pytest.mark.parametrize("success", [False, 0])
def test_known_pinstorage_key_generation_failure_does_not_warn(success, caplog):
record = {
"timestamp": "2026-06-17 15:31:02.014",
"key_generated": {
"success": success,
"key_id": "PinStorage_crossReboot_key",
"uid": 1001,
},
}
with caplog.at_level(logging.WARNING):
_run_security_heuristics([record])
assert "Failed key generation detected" not in caplog.text
timeline_event = SecurityEvent().serialize(record)
assert timeline_event["event"] == "key_generated"
assert "Key generation failed: PinStorage_crossReboot_key" in timeline_event["data"]
@pytest.mark.parametrize(
("key_id", "uid"),
[
("PinStorage_crossReboot_key", 10_000),
("another_key", 1001),
],
)
def test_other_key_generation_failures_still_warn(key_id, uid, caplog):
with caplog.at_level(logging.WARNING):
_run_security_heuristics(
[
{
"timestamp": "2026-06-17 15:31:02.014",
"key_generated": {
"success": False,
"key_id": key_id,
"uid": uid,
},
}
]
)
assert f"Failed key generation detected for key_id: {key_id}" in caplog.text
def test_cert_authority_installed_raises_medium_alert_without_indicators():
alerts = _run_security_heuristics(
[
-55
View File
@@ -7,11 +7,8 @@ import logging
from pathlib import Path
import pytest
from click.testing import CliRunner
from mvt.android.cli import check_androidqf
from mvt.android.modules.androidqf.aqf_packages import AQFPackages
from mvt.android.modules.androidqf import aqf_packages as aqf_packages_module
from mvt.common.module import run_module
from ..utils import get_android_androidqf, list_files
@@ -135,55 +132,3 @@ class TestAndroidqfPackages:
possible_detected_app[0].matched_indicator.value
== "c7e56178748be1441370416d4c10e34817ea0c961eb636c8e9d98e0fd79bf730"
)
def test_virustotal_delays_after_missing_result(self, monkeypatch):
lookups = []
sleeps = []
def fake_virustotal_lookup(file_hash):
lookups.append(file_hash)
if file_hash == "missing_hash":
return None
return {
"attributes": {
"last_analysis_stats": {"malicious": 1},
"last_analysis_results": {"engine": {}},
}
}
monkeypatch.setattr(
aqf_packages_module, "virustotal_lookup", fake_virustotal_lookup
)
monkeypatch.setattr(aqf_packages_module.time, "sleep", sleeps.append)
module = AQFPackages(
module_options={"virustotal": True, "virustotal_delay": 16},
results=[
{
"name": "org.example",
"installer": "com.android.vending",
"disabled": False,
"system": False,
"files": [
{"path": "/data/app/missing.apk", "sha256": "missing_hash"},
{"path": "/data/app/found.apk", "sha256": "found_hash"},
],
}
],
)
module.check_indicators()
assert lookups == ["missing_hash", "found_hash"]
assert sleeps == [16]
assert module.results[0]["files"][1]["virustotal"] == "1/1"
assert len(module.alertstore.alerts) == 1
def test_check_androidqf_rejects_negative_virustotal_delay(data_path):
runner = CliRunner()
result = runner.invoke(check_androidqf, ["--delay", "-1", data_path])
assert result.exit_code == 2
assert "Invalid value for '--delay'" in result.output
-2
View File
@@ -35,7 +35,6 @@ def indicators_factory(indicator_file):
domains=[],
emails=[],
file_names=[],
file_paths=[],
processes=[],
app_ids=[],
app_cert_hashes=[],
@@ -48,7 +47,6 @@ def indicators_factory(indicator_file):
ind.ioc_collections[0]["domains"].extend(domains)
ind.ioc_collections[0]["emails"].extend(emails)
ind.ioc_collections[0]["file_names"].extend(file_names)
ind.ioc_collections[0]["file_paths"].extend(file_paths)
ind.ioc_collections[0]["processes"].extend(processes)
ind.ioc_collections[0]["app_ids"].extend(app_ids)
ind.ioc_collections[0]["android_property_names"].extend(android_property_names)
-59
View File
@@ -1,59 +0,0 @@
# 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 mvt.common.module import run_module
from mvt.ios.modules.fs.shutdownlog import ShutdownLog
def _shutdown_log_entry(pid: int, client: str, timestamp: int) -> str:
return (
f"remaining client pid: {pid} ({client})\n"
f"SIGTERM: [{timestamp}]\n"
)
class TestShutdownLog:
def test_discovers_rotated_shutdown_logs(self, tmp_path):
diagnostics_path = tmp_path / "private/var/db/diagnostics"
diagnostics_path.mkdir(parents=True)
(diagnostics_path / "shutdown.log").write_text(
_shutdown_log_entry(100, "/usr/libexec/first", 1_700_000_000),
encoding="utf-8",
)
(diagnostics_path / "shutdown.0.log").write_text(
_shutdown_log_entry(200, "/usr/libexec/second", 1_700_000_001),
encoding="utf-8",
)
module = ShutdownLog(target_path=str(tmp_path))
run_module(module)
assert {result["client"] for result in module.results} == {
"/usr/libexec/first",
"/usr/libexec/second",
}
def test_file_path_indicator_matches_client_with_trailing_uuid(
self, indicators_factory
):
executable_path = "/usr/sbin/filecoordinationd"
client = f"{executable_path}/123e4567-e89b-12d3-a456-426614174000"
module = ShutdownLog(
results=[
{
"isodate": "2023-11-14 22:13:20.000000",
"pid": "100",
"client": client,
"delay": 0.0,
"times_delayed": 0,
}
]
)
module.indicators = indicators_factory(file_paths=[executable_path])
module.check_indicators()
assert len(module.alertstore.alerts) == 1
assert module.alertstore.alerts[0].matched_indicator.value == executable_path
-17
View File
@@ -3,8 +3,6 @@
# Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/
import shutil
from click.testing import CliRunner
from mvt.ios.cli import check_backup
@@ -18,18 +16,3 @@ class TestCheckBackupCommand:
path = get_ios_backup_folder()
result = runner.invoke(check_backup, [path])
assert result.exit_code == 0
def test_check_finds_backup_in_subfolder(self, tmp_path, caplog):
runner = CliRunner()
backup_path = tmp_path / "MobileSync" / "Backup" / "device-id"
shutil.copytree(get_ios_backup_folder(), backup_path)
result = runner.invoke(check_backup, [str(backup_path.parent)])
assert result.exit_code == 0
assert f"Found iTunes backup in subfolder: {backup_path}" in caplog.text
def test_check_rejects_non_backup_folder(self, tmp_path, caplog):
runner = CliRunner()
result = runner.invoke(check_backup, [str(tmp_path)])
assert result.exit_code == 1
assert "does not appear to be an iTunes backup folder" in caplog.text
-2
View File
@@ -63,8 +63,6 @@ def test_load_module_appears_only_for_supported_cli_command(tmp_path):
def test_module_option_runs_supported_custom_module(tmp_path):
(tmp_path / "Manifest.db").touch()
(tmp_path / "Info.plist").touch()
module_path = _write_custom_module(
tmp_path / "custom.py",
"CustomRunModule",