1
mirror of https://github.com/mvt-project/mvt synced 2025-11-13 01:37:36 +01:00

Compare commits

..

1 Commits

Author SHA1 Message Date
Donncha Ó Cearbhaill
612eb50ff2 Fix bug where default values were dropped when parsing protobuf tombstones 2025-03-11 13:48:04 +01:00
44 changed files with 213 additions and 1129 deletions

View File

@@ -1,11 +0,0 @@
# To get started with Dependabot version updates, you'll need to specify which
# package ecosystems to update and where the package manifests are located.
# Please see the documentation for all configuration options:
# https://docs.github.com/code-security/dependabot/dependabot-version-updates/configuration-options-for-the-dependabot.yml-file
version: 2
updates:
- package-ecosystem: "pip" # See documentation for possible values
directory: "/" # Location of package manifests
schedule:
interval: "weekly"

View File

@@ -12,7 +12,7 @@ jobs:
strategy:
fail-fast: false
matrix:
python-version: ['3.10', '3.11', '3.12', '3.13']
python-version: ['3.8', '3.9', '3.10'] # , '3.11']
steps:
- uses: actions/checkout@v4
@@ -35,4 +35,4 @@ jobs:
if: github.event_name == 'pull_request'
with:
pytest-coverage-path: ./pytest-coverage.txt
junitxml-path: ./pytest.xml
junitxml-path: ./pytest.xml

View File

@@ -103,7 +103,7 @@ RUN git clone https://github.com/libimobiledevice/usbmuxd && cd usbmuxd \
# Create main image
FROM ubuntu:24.04 as main
FROM ubuntu:22.04 as main
LABEL org.opencontainers.image.url="https://mvt.re"
LABEL org.opencontainers.image.documentation="https://docs.mvt.re"
@@ -135,7 +135,8 @@ COPY --from=build-usbmuxd /build /
COPY . mvt/
RUN apt-get update \
&& apt-get install -y git python3-pip \
&& PIP_NO_CACHE_DIR=1 pip3 install --break-system-packages ./mvt \
&& PIP_NO_CACHE_DIR=1 pip3 install --upgrade pip \
&& PIP_NO_CACHE_DIR=1 pip3 install ./mvt \
&& apt-get remove -y python3-pip git && apt-get autoremove -y \
&& rm -rf /var/lib/apt/lists/* \
&& rm -rf mvt

View File

@@ -23,7 +23,7 @@ install:
python3 -m pip install --upgrade -e .
test-requirements:
python3 -m pip install --upgrade --group dev
python3 -m pip install --upgrade -r test-requirements.txt
generate-proto-parsers:
# Generate python parsers for protobuf files

View File

@@ -1,5 +1,5 @@
mkdocs==1.6.1
mkdocs-autorefs==1.4.3
mkdocs-material==9.6.20
mkdocs-autorefs==1.2.0
mkdocs-material==9.5.42
mkdocs-material-extensions==1.3.1
mkdocstrings==0.30.1
mkdocstrings==0.23.0

View File

@@ -1,11 +1,13 @@
[project]
name = "mvt"
dynamic = ["version"]
authors = [{ name = "Claudio Guarnieri", email = "nex@nex.sx" }]
authors = [
{name = "Claudio Guarnieri", email = "nex@nex.sx"}
]
maintainers = [
{ name = "Etienne Maynier", email = "tek@randhome.io" },
{ name = "Donncha Ó Cearbhaill", email = "donncha.ocearbhaill@amnesty.org" },
{ name = "Rory Flynn", email = "rory.flynn@amnesty.org" },
{name = "Etienne Maynier", email = "tek@randhome.io"},
{name = "Donncha Ó Cearbhaill", email = "donncha.ocearbhaill@amnesty.org"},
{name = "Rory Flynn", email = "rory.flynn@amnesty.org"}
]
description = "Mobile Verification Toolkit"
readme = "README.md"
@@ -14,61 +16,48 @@ classifiers = [
"Development Status :: 5 - Production/Stable",
"Intended Audience :: Information Technology",
"Operating System :: OS Independent",
"Programming Language :: Python",
"Programming Language :: Python"
]
dependencies = [
"click==8.2.1",
"rich==14.1.0",
"tld==0.13.1",
"requests==2.32.4",
"simplejson==3.20.1",
"packaging==25.0",
"appdirs==1.4.4",
"iOSbackup==0.9.925",
"adb-shell[usb]==0.4.4",
"libusb1==3.3.1",
"cryptography==45.0.6",
"PyYAML>=6.0.2",
"pyahocorasick==2.2.0",
"betterproto==1.2.5",
"pydantic==2.11.7",
"pydantic-settings==2.10.1",
"NSKeyedUnArchiver==1.5.2",
"python-dateutil==2.9.0.post0",
"tzdata==2025.2",
"click >=8.1.3",
"rich >=12.6.0",
"tld >=0.12.6",
"requests >=2.28.1",
"simplejson >=3.17.6",
"packaging >=21.3",
"appdirs >=1.4.4",
"iOSbackup >=0.9.923",
"adb-shell[usb] >=0.4.3",
"libusb1 >=3.0.0",
"cryptography >=42.0.5",
"pyyaml >=6.0",
"pyahocorasick >= 2.0.0",
"betterproto >=1.2.0",
"pydantic >= 2.10.0",
"pydantic-settings >= 2.7.0",
'backports.zoneinfo; python_version < "3.9"',
]
requires-python = ">= 3.10"
requires-python = ">= 3.8"
[project.urls]
homepage = "https://docs.mvt.re/en/latest/"
repository = "https://github.com/mvt-project/mvt"
[project.scripts]
mvt-ios = "mvt.ios:cli"
mvt-android = "mvt.android:cli"
[dependency-groups]
dev = [
"requests>=2.31.0",
"pytest>=7.4.3",
"pytest-cov>=4.1.0",
"pytest-github-actions-annotate-failures>=0.2.0",
"pytest-mock>=3.14.0",
"stix2>=3.0.1",
"ruff>=0.1.6",
"mypy>=1.7.1",
"betterproto[compiler]",
]
mvt-ios = "mvt.ios:cli"
mvt-android = "mvt.android:cli"
[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"
[tool.coverage.run]
omit = ["tests/*"]
omit = [
"tests/*",
]
[tool.coverage.html]
directory = "htmlcov"
directory= "htmlcov"
[tool.mypy]
install_types = true
@@ -78,13 +67,15 @@ packages = "src"
[tool.pytest.ini_options]
addopts = "-ra -q --cov=mvt --cov-report html --junitxml=pytest.xml --cov-report=term-missing:skip-covered"
testpaths = ["tests"]
testpaths = [
"tests"
]
[tool.ruff.lint]
select = ["C90", "E", "F", "W"] # flake8 default set
select = ["C90", "E", "F", "W"] # flake8 default set
ignore = [
"E501", # don't enforce line length violations
"C901", # complex-structure
"E501", # don't enforce line length violations
"C901", # complex-structure
# These were previously ignored but don't seem to be required:
# "E265", # no-space-after-block-comment
@@ -96,14 +87,14 @@ ignore = [
]
[tool.ruff.lint.per-file-ignores]
"__init__.py" = ["F401"] # unused-import
"__init__.py" = ["F401"] # unused-import
[tool.ruff.lint.mccabe]
max-complexity = 10
[tool.setuptools]
include-package-data = true
package-dir = { "" = "src" }
package-dir = {"" = "src"}
[tool.setuptools.packages.find]
where = ["src"]
@@ -112,4 +103,4 @@ where = ["src"]
mvt = ["ios/data/*.json"]
[tool.setuptools.dynamic]
version = { attr = "mvt.common.version.MVT_VERSION" }
version = {attr = "mvt.common.version.MVT_VERSION"}

View File

@@ -4,7 +4,6 @@
# https://license.mvt.re/1.1/
import base64
import binascii
import hashlib
from .artifact import AndroidArtifact
@@ -90,16 +89,11 @@ class DumpsysADBArtifact(AndroidArtifact):
else:
key_base64, user = user_key, b""
try:
key_raw = base64.b64decode(key_base64)
key_fingerprint = hashlib.md5(key_raw).hexdigest().upper()
key_fingerprint_colon = ":".join(
[key_fingerprint[i : i + 2] for i in range(0, len(key_fingerprint), 2)]
)
except binascii.Error:
# Impossible to parse base64
key_fingerprint_colon = ""
key_raw = base64.b64decode(key_base64)
key_fingerprint = hashlib.md5(key_raw).hexdigest().upper()
key_fingerprint_colon = ":".join(
[key_fingerprint[i : i + 2] for i in range(0, len(key_fingerprint), 2)]
)
return {
"user": user.decode("utf-8"),
"fingerprint": key_fingerprint_colon,

View File

@@ -1,186 +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 typing import Any
from .artifact import AndroidArtifact
SUSPICIOUS_MOUNT_POINTS = [
"/system",
"/vendor",
"/product",
"/system_ext",
]
SUSPICIOUS_OPTIONS = [
"rw",
"remount",
"noatime",
"nodiratime",
]
ALLOWLIST_NOATIME = [
"/system_dlkm",
"/system_ext",
"/product",
"/vendor",
"/vendor_dlkm",
]
class Mounts(AndroidArtifact):
"""
This artifact parses mount information from /proc/mounts or similar mount data.
It can detect potentially suspicious mount configurations that may indicate
a rooted or compromised device.
"""
def parse(self, entry: str) -> None:
"""
Parse mount information from the provided entry.
Examples:
/dev/block/bootdevice/by-name/system /system ext4 ro,seclabel,relatime 0 0
/dev/block/dm-12 on / type ext4 (ro,seclabel,noatime)
"""
self.results: list[dict[str, Any]] = []
for line in entry.splitlines():
line = line.strip()
if not line:
continue
device = None
mount_point = None
filesystem_type = None
mount_options = ""
if " on " in line and " type " in line:
try:
# Format: device on mount_point type filesystem_type (options)
device_part, rest = line.split(" on ", 1)
device = device_part.strip()
# Split by 'type' to get mount_point and filesystem info
mount_part, fs_part = rest.split(" type ", 1)
mount_point = mount_part.strip()
# Parse filesystem and options
if "(" in fs_part and fs_part.endswith(")"):
# Format: filesystem_type (options)
fs_and_opts = fs_part.strip()
paren_idx = fs_and_opts.find("(")
filesystem_type = fs_and_opts[:paren_idx].strip()
mount_options = fs_and_opts[paren_idx + 1 : -1].strip()
else:
# No options in parentheses, just filesystem type
filesystem_type = fs_part.strip()
mount_options = ""
# Skip if we don't have essential info
if not device or not mount_point or not filesystem_type:
continue
# Parse options into list
options_list = (
[opt.strip() for opt in mount_options.split(",") if opt.strip()]
if mount_options
else []
)
# Check if it's a system partition
is_system_partition = mount_point in SUSPICIOUS_MOUNT_POINTS or any(
mount_point.startswith(sp) for sp in SUSPICIOUS_MOUNT_POINTS
)
# Check if it's mounted read-write
is_read_write = "rw" in options_list
mount_entry = {
"device": device,
"mount_point": mount_point,
"filesystem_type": filesystem_type,
"mount_options": mount_options,
"options_list": options_list,
"is_system_partition": is_system_partition,
"is_read_write": is_read_write,
}
self.results.append(mount_entry)
except ValueError:
# If parsing fails, skip this line
continue
else:
# Skip lines that don't match expected format
continue
def check_indicators(self) -> None:
"""
Check for suspicious mount configurations that may indicate root access
or other security concerns.
"""
system_rw_mounts = []
suspicious_mounts = []
for mount in self.results:
mount_point = mount["mount_point"]
options = mount["options_list"]
# Check for system partitions mounted as read-write
if mount["is_system_partition"] and mount["is_read_write"]:
system_rw_mounts.append(mount)
if mount_point == "/system":
self.log.warning(
"Root detected /system partition is mounted as read-write (rw). "
)
else:
self.log.warning(
"System partition %s is mounted as read-write (rw). This may indicate system modifications.",
mount_point,
)
# Check for other suspicious mount options
suspicious_opts = [opt for opt in options if opt in SUSPICIOUS_OPTIONS]
if suspicious_opts and mount["is_system_partition"]:
if (
"noatime" in mount["mount_options"]
and mount["mount_point"] in ALLOWLIST_NOATIME
):
continue
suspicious_mounts.append(mount)
self.log.warning(
"Suspicious mount options found for %s: %s",
mount_point,
", ".join(suspicious_opts),
)
# Log interesting mount information
if mount_point == "/data" or mount_point.startswith("/sdcard"):
self.log.info(
"Data partition: %s mounted as %s with options: %s",
mount_point,
mount["filesystem_type"],
mount["mount_options"],
)
self.log.info("Parsed %d mount entries", len(self.results))
# Check indicators if available
if not self.indicators:
return
for mount in self.results:
# Check if any mount points match indicators
ioc = self.indicators.check_file_path(mount.get("mount_point", ""))
if ioc:
mount["matched_indicator"] = ioc
self.detected.append(mount)
# Check device paths for indicators
ioc = self.indicators.check_file_path(mount.get("device", ""))
if ioc:
mount["matched_indicator"] = ioc
self.detected.append(mount)

View File

@@ -51,6 +51,11 @@ ANDROID_DANGEROUS_SETTINGS = [
"key": "send_action_app_error",
"safe_value": "1",
},
{
"description": "enabled installation of non Google Play apps",
"key": "install_non_market_apps",
"safe_value": "0",
},
{
"description": "enabled accessibility services",
"key": "accessibility_enabled",

View File

@@ -8,7 +8,6 @@ from typing import List, Optional, Union
import pydantic
import betterproto
from dateutil import parser
from mvt.common.utils import convert_datetime_to_iso
from mvt.android.parsers.proto.tombstone import Tombstone
@@ -53,7 +52,7 @@ class TombstoneCrashResult(pydantic.BaseModel):
file_name: str
file_timestamp: str # We store the timestamp as a string to avoid timezone issues
build_fingerprint: str
revision: str
revision: int
arch: Optional[str] = None
timestamp: str # We store the timestamp as a string to avoid timezone issues
process_uptime: Optional[int] = None
@@ -70,7 +69,7 @@ class TombstoneCrashResult(pydantic.BaseModel):
class TombstoneCrashArtifact(AndroidArtifact):
"""
""" "
Parser for Android tombstone crash files.
This parser can parse both text and protobuf tombstone crash files.
@@ -121,7 +120,9 @@ class TombstoneCrashArtifact(AndroidArtifact):
def parse_protobuf(
self, file_name: str, file_timestamp: datetime.datetime, data: bytes
) -> None:
"""Parse Android tombstone crash files from a protobuf object."""
"""
Parse Android tombstone crash files from a protobuf object.
"""
tombstone_pb = Tombstone().parse(data)
tombstone_dict = tombstone_pb.to_dict(
betterproto.Casing.SNAKE, include_default_values=True
@@ -142,23 +143,21 @@ class TombstoneCrashArtifact(AndroidArtifact):
def parse(
self, file_name: str, file_timestamp: datetime.datetime, content: bytes
) -> None:
"""Parse text Android tombstone crash files."""
"""
Parse text Android tombstone crash files.
"""
# Split the tombstone file into a dictonary
tombstone_dict = {
"file_name": file_name,
"file_timestamp": convert_datetime_to_iso(file_timestamp),
}
lines = content.decode("utf-8").splitlines()
for line_num, line in enumerate(lines, 1):
for line in lines:
if not line.strip() or TOMBSTONE_DELIMITER in line:
continue
try:
for key, destination_key in TOMBSTONE_TEXT_KEY_MAPPINGS.items():
if self._parse_tombstone_line(
line, key, destination_key, tombstone_dict
):
break
except Exception as e:
raise ValueError(f"Error parsing line {line_num}: {str(e)}")
for key, destination_key in TOMBSTONE_TEXT_KEY_MAPPINGS.items():
self._parse_tombstone_line(line, key, destination_key, tombstone_dict)
# Validate the tombstone and add it to the results
tombstone = TombstoneCrashResult.model_validate(tombstone_dict)
@@ -168,7 +167,7 @@ class TombstoneCrashArtifact(AndroidArtifact):
self, line: str, key: str, destination_key: str, tombstone: dict
) -> bool:
if not line.startswith(f"{key}"):
return False
return None
if key == "pid":
return self._load_pid_line(line, tombstone)
@@ -187,7 +186,7 @@ class TombstoneCrashArtifact(AndroidArtifact):
raise ValueError(f"Expected key {key}, got {line_key}")
value_clean = value.strip().strip("'")
if destination_key == "uid":
if destination_key in ["uid", "revision"]:
tombstone[destination_key] = int(value_clean)
elif destination_key == "process_uptime":
# eg. "Process uptime: 40s"
@@ -200,50 +199,51 @@ class TombstoneCrashArtifact(AndroidArtifact):
return True
def _load_pid_line(self, line: str, tombstone: dict) -> bool:
try:
parts = line.split(" >>> ") if " >>> " in line else line.split(">>>")
process_info = parts[0]
pid_part, tid_part, name_part = [part.strip() for part in line.split(",")]
# Parse pid, tid, name from process info
info_parts = [p.strip() for p in process_info.split(",")]
for info in info_parts:
key, value = info.split(":", 1)
key = key.strip()
value = value.strip()
pid_key, pid_value = pid_part.split(":", 1)
if pid_key != "pid":
raise ValueError(f"Expected key pid, got {pid_key}")
pid_value = int(pid_value.strip())
if key == "pid":
tombstone["pid"] = int(value)
elif key == "tid":
tombstone["tid"] = int(value)
elif key == "name":
tombstone["process_name"] = value
tid_key, tid_value = tid_part.split(":", 1)
if tid_key != "tid":
raise ValueError(f"Expected key tid, got {tid_key}")
tid_value = int(tid_value.strip())
# Extract binary path if it exists
if len(parts) > 1:
tombstone["binary_path"] = parts[1].strip().rstrip(" <")
name_key, name_value = name_part.split(":", 1)
if name_key != "name":
raise ValueError(f"Expected key name, got {name_key}")
name_value = name_value.strip()
process_name, binary_path = self._parse_process_name(name_value, tombstone)
return True
tombstone["pid"] = pid_value
tombstone["tid"] = tid_value
tombstone["process_name"] = process_name
tombstone["binary_path"] = binary_path
return True
except Exception as e:
raise ValueError(f"Failed to parse PID line: {str(e)}")
def _parse_process_name(self, process_name_part, tombstone: dict) -> bool:
process_name, process_path = process_name_part.split(">>>")
process_name = process_name.strip()
binary_path = process_path.strip().split(" ")[0]
return process_name, binary_path
def _load_signal_line(self, line: str, tombstone: dict) -> bool:
signal_part, code_part = map(str.strip, line.split(",")[:2])
signal, code, _ = [part.strip() for part in line.split(",", 2)]
signal = signal.split("signal ")[1]
signal_code, signal_name = signal.split(" ")
signal_name = signal_name.strip("()")
def parse_part(part: str, prefix: str) -> tuple[int, str]:
match = part.split(prefix)[1]
number = int(match.split()[0])
name = match.split("(")[1].split(")")[0] if "(" in match else "UNKNOWN"
return number, name
signal_number, signal_name = parse_part(signal_part, "signal ")
code_number, code_name = parse_part(code_part, "code ")
code_part = code.split("code ")[1]
code_number, code_name = code_part.split(" ")
code_name = code_name.strip("()")
tombstone["signal_info"] = {
"code": code_number,
"code": int(code_number),
"code_name": code_name,
"name": signal_name,
"number": signal_number,
"number": int(signal_code),
}
return True
@@ -254,7 +254,13 @@ class TombstoneCrashArtifact(AndroidArtifact):
@staticmethod
def _parse_timestamp_string(timestamp: str) -> str:
timestamp_parsed = parser.parse(timestamp)
timestamp_date, timezone = timestamp.split("+")
# Truncate microseconds before parsing
timestamp_without_micro = timestamp_date.split(".")[0] + "+" + timezone
timestamp_parsed = datetime.datetime.strptime(
timestamp_without_micro, "%Y-%m-%d %H:%M:%S%z"
)
# HACK: Swap the local timestamp to UTC, so keep the original time and avoid timezone conversion.
local_timestamp = timestamp_parsed.replace(tzinfo=datetime.timezone.utc)
return convert_datetime_to_iso(local_timestamp)

View File

@@ -31,8 +31,6 @@ from mvt.common.help import (
HELP_MSG_HASHES,
HELP_MSG_CHECK_IOCS,
HELP_MSG_STIX2,
HELP_MSG_DISABLE_UPDATE_CHECK,
HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK,
)
from mvt.common.logo import logo
from mvt.common.updates import IndicatorsUpdates
@@ -55,37 +53,12 @@ log = logging.getLogger("mvt")
CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])
def _get_disable_flags(ctx):
"""Helper function to safely get disable flags from context."""
if ctx.obj is None:
return False, False
return (
ctx.obj.get("disable_version_check", False),
ctx.obj.get("disable_indicator_check", False),
)
# ==============================================================================
# Main
# ==============================================================================
@click.group(invoke_without_command=False)
@click.option(
"--disable-update-check", is_flag=True, help=HELP_MSG_DISABLE_UPDATE_CHECK
)
@click.option(
"--disable-indicator-update-check",
is_flag=True,
help=HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK,
)
@click.pass_context
def cli(ctx, disable_update_check, disable_indicator_update_check):
ctx.ensure_object(dict)
ctx.obj["disable_version_check"] = disable_update_check
ctx.obj["disable_indicator_check"] = disable_indicator_update_check
logo(
disable_version_check=disable_update_check,
disable_indicator_check=disable_indicator_update_check,
)
def cli():
logo()
# ==============================================================================
@@ -193,8 +166,6 @@ def check_adb(
module_name=module,
serial=serial,
module_options=module_options,
disable_version_check=_get_disable_flags(ctx)[0],
disable_indicator_check=_get_disable_flags(ctx)[1],
)
if list_modules:
@@ -241,8 +212,6 @@ def check_bugreport(ctx, iocs, output, list_modules, module, verbose, bugreport_
ioc_files=iocs,
module_name=module,
hashes=True,
disable_version_check=_get_disable_flags(ctx)[0],
disable_indicator_check=_get_disable_flags(ctx)[1],
)
if list_modules:
@@ -305,8 +274,6 @@ def check_backup(
"interactive": not non_interactive,
"backup_password": cli_load_android_backup_password(log, backup_password),
},
disable_version_check=_get_disable_flags(ctx)[0],
disable_indicator_check=_get_disable_flags(ctx)[1],
)
if list_modules:
@@ -371,8 +338,6 @@ def check_androidqf(
"interactive": not non_interactive,
"backup_password": cli_load_android_backup_password(log, backup_password),
},
disable_version_check=_get_disable_flags(ctx)[0],
disable_indicator_check=_get_disable_flags(ctx)[1],
)
if list_modules:
@@ -407,13 +372,7 @@ def check_androidqf(
@click.argument("FOLDER", type=click.Path(exists=True))
@click.pass_context
def check_iocs(ctx, iocs, list_modules, module, folder):
cmd = CmdCheckIOCS(
target_path=folder,
ioc_files=iocs,
module_name=module,
disable_version_check=_get_disable_flags(ctx)[0],
disable_indicator_check=_get_disable_flags(ctx)[1],
)
cmd = CmdCheckIOCS(target_path=folder, ioc_files=iocs, module_name=module)
cmd.modules = BACKUP_MODULES + ADB_MODULES + BUGREPORT_MODULES
if list_modules:

View File

@@ -22,8 +22,6 @@ class CmdAndroidCheckADB(Command):
module_name: Optional[str] = None,
serial: Optional[str] = None,
module_options: Optional[dict] = None,
disable_version_check: bool = False,
disable_indicator_check: bool = False,
) -> None:
super().__init__(
target_path=target_path,
@@ -33,8 +31,6 @@ class CmdAndroidCheckADB(Command):
serial=serial,
module_options=module_options,
log=log,
disable_version_check=disable_version_check,
disable_indicator_check=disable_indicator_check,
)
self.name = "check-adb"

View File

@@ -26,8 +26,6 @@ class CmdAndroidCheckAndroidQF(Command):
serial: Optional[str] = None,
module_options: Optional[dict] = None,
hashes: bool = False,
disable_version_check: bool = False,
disable_indicator_check: bool = False,
) -> None:
super().__init__(
target_path=target_path,
@@ -38,8 +36,6 @@ class CmdAndroidCheckAndroidQF(Command):
module_options=module_options,
hashes=hashes,
log=log,
disable_version_check=disable_version_check,
disable_indicator_check=disable_indicator_check,
)
self.name = "check-androidqf"

View File

@@ -36,8 +36,6 @@ class CmdAndroidCheckBackup(Command):
serial: Optional[str] = None,
module_options: Optional[dict] = None,
hashes: bool = False,
disable_version_check: bool = False,
disable_indicator_check: bool = False,
) -> None:
super().__init__(
target_path=target_path,
@@ -48,8 +46,6 @@ class CmdAndroidCheckBackup(Command):
module_options=module_options,
hashes=hashes,
log=log,
disable_version_check=disable_version_check,
disable_indicator_check=disable_indicator_check,
)
self.name = "check-backup"

View File

@@ -27,8 +27,6 @@ class CmdAndroidCheckBugreport(Command):
serial: Optional[str] = None,
module_options: Optional[dict] = None,
hashes: bool = False,
disable_version_check: bool = False,
disable_indicator_check: bool = False,
) -> None:
super().__init__(
target_path=target_path,
@@ -39,8 +37,6 @@ class CmdAndroidCheckBugreport(Command):
module_options=module_options,
hashes=hashes,
log=log,
disable_version_check=disable_version_check,
disable_indicator_check=disable_indicator_check,
)
self.name = "check-bugreport"

View File

@@ -107,7 +107,8 @@ class Packages(AndroidExtraction):
result["matched_indicator"] = ioc
self.detected.append(result)
def check_virustotal(self, packages: list) -> None:
@staticmethod
def check_virustotal(packages: list) -> None:
hashes = []
for package in packages:
for file in package.get("files", []):
@@ -142,15 +143,8 @@ class Packages(AndroidExtraction):
for package in packages:
for file in package.get("files", []):
if "package_name" in package:
row = [package["package_name"], file["path"]]
elif "name" in package:
row = [package["name"], file["path"]]
else:
self.log.error(
f"Package {package} has no name or package_name. packages.json or apks.json is malformed"
)
continue
row = [package["package_name"], file["path"]]
if file["sha256"] in detections:
detection = detections[file["sha256"]]
positives = detection.split("/")[0]

View File

@@ -19,8 +19,6 @@ from .processes import Processes
from .settings import Settings
from .sms import SMS
from .files import Files
from .root_binaries import RootBinaries
from .mounts import Mounts
ANDROIDQF_MODULES = [
DumpsysActivities,
@@ -39,6 +37,4 @@ ANDROIDQF_MODULES = [
SMS,
DumpsysPackages,
Files,
RootBinaries,
Mounts,
]

View File

@@ -1,74 +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
import json
from typing import Optional
from mvt.android.artifacts.mounts import Mounts as MountsArtifact
from .base import AndroidQFModule
class Mounts(MountsArtifact, AndroidQFModule):
"""This module extracts and analyzes mount information from AndroidQF acquisitions."""
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[list] = 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.results = []
def run(self) -> None:
"""
Run the mounts analysis module.
This module looks for mount information files collected by androidqf
and analyzes them for suspicious configurations, particularly focusing
on detecting root access indicators like /system mounted as read-write.
"""
mount_files = self._get_files_by_pattern("*/mounts.json")
if not mount_files:
self.log.info("No mount information file found")
return
self.log.info("Found mount information file: %s", mount_files[0])
try:
data = self._get_file_content(mount_files[0]).decode(
"utf-8", errors="replace"
)
except Exception as exc:
self.log.error("Failed to read mount information file: %s", exc)
return
# Parse the mount data
try:
json_data = json.loads(data)
if isinstance(json_data, list):
# AndroidQF format: array of strings like
# "/dev/block/dm-12 on / type ext4 (ro,seclabel,noatime)"
mount_content = "\n".join(json_data)
self.parse(mount_content)
except Exception as exc:
self.log.error("Failed to parse mount information: %s", exc)
return
self.log.info("Extracted a total of %d mount entries", len(self.results))

View File

@@ -1,121 +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 json
import logging
from typing import Optional
from .base import AndroidQFModule
class RootBinaries(AndroidQFModule):
"""This module analyzes root_binaries.json for root binaries found by androidqf."""
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[list] = 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: dict) -> dict:
return {
"timestamp": record.get("timestamp"),
"module": self.__class__.__name__,
"event": "root_binary_found",
"data": f"Root binary found: {record['path']} (binary: {record['binary_name']})",
}
def check_indicators(self) -> None:
"""Check for indicators of device rooting."""
if not self.results:
return
# All found root binaries are considered indicators of rooting
for result in self.results:
self.log.warning(
'Found root binary "%s" at path "%s"',
result["binary_name"],
result["path"],
)
self.detected.append(result)
if self.detected:
self.log.warning(
"Device shows signs of rooting with %d root binaries found",
len(self.detected),
)
def run(self) -> None:
"""Run the root binaries analysis."""
root_binaries_files = self._get_files_by_pattern("*/root_binaries.json")
if not root_binaries_files:
self.log.info("No root_binaries.json file found")
return
rawdata = self._get_file_content(root_binaries_files[0]).decode(
"utf-8", errors="ignore"
)
try:
root_binary_paths = json.loads(rawdata)
except json.JSONDecodeError as e:
self.log.error("Failed to parse root_binaries.json: %s", e)
return
if not isinstance(root_binary_paths, list):
self.log.error("Expected root_binaries.json to contain a list of paths")
return
# Known root binary names that might be found and their descriptions
# This maps the binary name to a human-readable description
known_root_binaries = {
"su": "SuperUser binary",
"busybox": "BusyBox utilities",
"supersu": "SuperSU root management",
"Superuser.apk": "Superuser app",
"KingoUser.apk": "KingRoot app",
"SuperSu.apk": "SuperSU app",
"magisk": "Magisk root framework",
"magiskhide": "Magisk hide utility",
"magiskinit": "Magisk init binary",
"magiskpolicy": "Magisk policy binary",
}
for path in root_binary_paths:
if not path or not isinstance(path, str):
continue
# Extract binary name from path
binary_name = path.split("/")[-1].lower()
# Check if this matches a known root binary by exact name match
description = "Unknown root binary"
for known_binary in known_root_binaries:
if binary_name == known_binary.lower():
description = known_root_binaries[known_binary]
break
result = {
"path": path.strip(),
"binary_name": binary_name,
"description": description,
}
self.results.append(result)
self.log.info("Found %d root binaries", len(self.results))

View File

@@ -231,7 +231,6 @@ def parse_sms_file(data):
entry.pop("mms_body")
body = entry.get("body", None)
message_links = None
if body:
message_links = check_for_links(entry["body"])

View File

@@ -22,8 +22,6 @@ class CmdCheckIOCS(Command):
module_name: Optional[str] = None,
serial: Optional[str] = None,
module_options: Optional[dict] = None,
disable_version_check: bool = False,
disable_indicator_check: bool = False,
) -> None:
super().__init__(
target_path=target_path,
@@ -33,8 +31,6 @@ class CmdCheckIOCS(Command):
serial=serial,
module_options=module_options,
log=log,
disable_version_check=disable_version_check,
disable_indicator_check=disable_indicator_check,
)
self.name = "check-iocs"
@@ -69,10 +65,6 @@ class CmdCheckIOCS(Command):
m = iocs_module.from_json(
file_path, log=logging.getLogger(iocs_module.__module__)
)
if not m:
log.warning("No result from this module, skipping it")
continue
if self.iocs.total_ioc_count > 0:
m.indicators = self.iocs
m.indicators.log = m.log

View File

@@ -32,8 +32,6 @@ class Command:
module_options: Optional[dict] = None,
hashes: bool = False,
log: logging.Logger = logging.getLogger(__name__),
disable_version_check: bool = False,
disable_indicator_check: bool = False,
) -> None:
self.name = ""
self.modules = []
@@ -44,8 +42,6 @@ class Command:
self.module_name = module_name
self.serial = serial
self.log = log
self.disable_version_check = disable_version_check
self.disable_indicator_check = disable_indicator_check
# This dictionary can contain options that will be passed down from
# the Command to all modules. This can for example be used to pass

View File

@@ -15,8 +15,6 @@ HELP_MSG_HASHES = "Generate hashes of all the files analyzed"
HELP_MSG_VERBOSE = "Verbose mode"
HELP_MSG_CHECK_IOCS = "Compare stored JSON results to provided indicators"
HELP_MSG_STIX2 = "Download public STIX2 indicators"
HELP_MSG_DISABLE_UPDATE_CHECK = "Disable MVT version update check"
HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK = "Disable indicators update check"
# IOS Specific
HELP_MSG_DECRYPT_BACKUP = "Decrypt an encrypted iTunes backup"

View File

@@ -12,85 +12,74 @@ from .updates import IndicatorsUpdates, MVTUpdates
from .version import MVT_VERSION
def check_updates(
disable_version_check: bool = False, disable_indicator_check: bool = False
) -> None:
def check_updates() -> None:
log = logging.getLogger("mvt")
# First we check for MVT version updates.
if not disable_version_check:
try:
mvt_updates = MVTUpdates()
latest_version = mvt_updates.check()
except (requests.exceptions.ConnectionError, requests.exceptions.Timeout):
try:
mvt_updates = MVTUpdates()
latest_version = mvt_updates.check()
except (requests.exceptions.ConnectionError, requests.exceptions.Timeout):
rich_print(
"\t\t[bold]Note: Could not check for MVT updates.[/bold] "
"You may be working offline. Please update MVT regularly."
)
except Exception as e:
log.error("Error encountered when trying to check latest MVT version: %s", e)
else:
if latest_version:
rich_print(
"\t\t[bold]Note: Could not check for MVT updates.[/bold] "
"You may be working offline. Please update MVT regularly."
f"\t\t[bold]Version {latest_version} is available! "
"Upgrade mvt with `pip3 install -U mvt` or with `pipx upgrade mvt`[/bold]"
)
except Exception as e:
log.error(
"Error encountered when trying to check latest MVT version: %s", e
)
else:
if latest_version:
rich_print(
f"\t\t[bold]Version {latest_version} is available! "
"Upgrade mvt with `pip3 install -U mvt` or with `pipx upgrade mvt`[/bold]"
)
# Then we check for indicators files updates.
if not disable_indicator_check:
ioc_updates = IndicatorsUpdates()
ioc_updates = IndicatorsUpdates()
# Before proceeding, we check if we have downloaded an indicators index.
# If not, there's no point in proceeding with the updates check.
if ioc_updates.get_latest_update() == 0:
rich_print(
"\t\t[bold]You have not yet downloaded any indicators, check "
"the `download-iocs` command![/bold]"
)
return
# Before proceeding, we check if we have downloaded an indicators index.
# If not, there's no point in proceeding with the updates check.
if ioc_updates.get_latest_update() == 0:
rich_print(
"\t\t[bold]You have not yet downloaded any indicators, check "
"the `download-iocs` command![/bold]"
)
return
# We only perform this check at a fixed frequency, in order to not
# overburden the user with too many lookups if the command is being run
# multiple times.
should_check, hours = ioc_updates.should_check()
if not should_check:
rich_print(
f"\t\tIndicators updates checked recently, next automatic check "
f"in {int(hours)} hours"
)
return
# We only perform this check at a fixed frequency, in order to not
# overburden the user with too many lookups if the command is being run
# multiple times.
should_check, hours = ioc_updates.should_check()
if not should_check:
rich_print(
f"\t\tIndicators updates checked recently, next automatic check "
f"in {int(hours)} hours"
)
return
try:
ioc_to_update = ioc_updates.check()
except (requests.exceptions.ConnectionError, requests.exceptions.Timeout):
try:
ioc_to_update = ioc_updates.check()
except (requests.exceptions.ConnectionError, requests.exceptions.Timeout):
rich_print(
"\t\t[bold]Note: Could not check for indicator updates.[/bold] "
"You may be working offline. Please update MVT indicators regularly."
)
except Exception as e:
log.error("Error encountered when trying to check latest MVT indicators: %s", e)
else:
if ioc_to_update:
rich_print(
"\t\t[bold]Note: Could not check for indicator updates.[/bold] "
"You may be working offline. Please update MVT indicators regularly."
)
except Exception as e:
log.error(
"Error encountered when trying to check latest MVT indicators: %s", e
"\t\t[bold]There are updates to your indicators files! "
"Run the `download-iocs` command to update![/bold]"
)
else:
if ioc_to_update:
rich_print(
"\t\t[bold]There are updates to your indicators files! "
"Run the `download-iocs` command to update![/bold]"
)
else:
rich_print("\t\tYour indicators files seem to be up to date.")
rich_print("\t\tYour indicators files seem to be up to date.")
def logo(
disable_version_check: bool = False, disable_indicator_check: bool = False
) -> None:
def logo() -> None:
rich_print("\n")
rich_print("\t[bold]MVT[/bold] - Mobile Verification Toolkit")
rich_print("\t\thttps://mvt.re")
rich_print(f"\t\tVersion: {MVT_VERSION}")
check_updates(disable_version_check, disable_indicator_check)
check_updates()
rich_print("\n")

View File

@@ -69,14 +69,10 @@ class MVTModule:
@classmethod
def from_json(cls, json_path: str, log: logging.Logger):
with open(json_path, "r", encoding="utf-8") as handle:
try:
results = json.load(handle)
if log:
log.info('Loaded %d results from "%s"', len(results), json_path)
return cls(results=results, log=log)
except json.decoder.JSONDecodeError as err:
log.error('Error to decode the json "%s" file: "%s"', json_path, err)
return None
results = json.load(handle)
if log:
log.info('Loaded %d results from "%s"', len(results), json_path)
return cls(results=results, log=log)
@classmethod
def get_slug(cls) -> str:

View File

@@ -24,11 +24,7 @@ INDICATORS_CHECK_FREQUENCY = 12
class MVTUpdates:
def check(self) -> str:
try:
res = requests.get(settings.PYPI_UPDATE_URL, timeout=5)
except requests.exceptions.RequestException as e:
log.error("Failed to check for updates, skipping updates: %s", e)
return ""
res = requests.get(settings.PYPI_UPDATE_URL, timeout=15)
data = res.json()
latest_version = data.get("info", {}).get("version", "")
@@ -97,12 +93,7 @@ class IndicatorsUpdates:
url = self.github_raw_url.format(
self.index_owner, self.index_repo, self.index_branch, self.index_path
)
try:
res = requests.get(url, timeout=5)
except requests.exceptions.RequestException as e:
log.error("Failed to retrieve indicators index from %s: %s", url, e)
return None
res = requests.get(url, timeout=15)
if res.status_code != 200:
log.error(
"Failed to retrieve indicators index located at %s (error %d)",
@@ -114,12 +105,7 @@ class IndicatorsUpdates:
return yaml.safe_load(res.content)
def download_remote_ioc(self, ioc_url: str) -> Optional[str]:
try:
res = requests.get(ioc_url, timeout=15)
except requests.exceptions.RequestException as e:
log.error("Failed to download indicators file from %s: %s", ioc_url, e)
return None
res = requests.get(ioc_url, timeout=15)
if res.status_code != 200:
log.error(
"Failed to download indicators file from %s (error %d)",
@@ -185,12 +171,7 @@ class IndicatorsUpdates:
file_commit_url = (
f"https://api.github.com/repos/{owner}/{repo}/commits?path={path}"
)
try:
res = requests.get(file_commit_url, timeout=5)
except requests.exceptions.RequestException as e:
log.error("Failed to get details about file %s: %s", file_commit_url, e)
return -1
res = requests.get(file_commit_url, timeout=15)
if res.status_code != 200:
log.error(
"Failed to get details about file %s (error %d)",

View File

@@ -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 = "2.6.1"
MVT_VERSION = "2.6.0"

View File

@@ -37,8 +37,6 @@ from mvt.common.help import (
HELP_MSG_CHECK_IOCS,
HELP_MSG_STIX2,
HELP_MSG_CHECK_IOS_BACKUP,
HELP_MSG_DISABLE_UPDATE_CHECK,
HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK,
)
from .cmd_check_backup import CmdIOSCheckBackup
from .cmd_check_fs import CmdIOSCheckFS
@@ -55,37 +53,12 @@ MVT_IOS_BACKUP_PASSWORD = "MVT_IOS_BACKUP_PASSWORD"
CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])
def _get_disable_flags(ctx):
"""Helper function to safely get disable flags from context."""
if ctx.obj is None:
return False, False
return (
ctx.obj.get("disable_version_check", False),
ctx.obj.get("disable_indicator_check", False),
)
# ==============================================================================
# Main
# ==============================================================================
@click.group(invoke_without_command=False)
@click.option(
"--disable-update-check", is_flag=True, help=HELP_MSG_DISABLE_UPDATE_CHECK
)
@click.option(
"--disable-indicator-update-check",
is_flag=True,
help=HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK,
)
@click.pass_context
def cli(ctx, disable_update_check, disable_indicator_update_check):
ctx.ensure_object(dict)
ctx.obj["disable_version_check"] = disable_update_check
ctx.obj["disable_indicator_check"] = disable_indicator_update_check
logo(
disable_version_check=disable_update_check,
disable_indicator_check=disable_indicator_update_check,
)
def cli():
logo()
# ==============================================================================
@@ -246,8 +219,6 @@ def check_backup(
module_name=module,
module_options=module_options,
hashes=hashes,
disable_version_check=_get_disable_flags(ctx)[0],
disable_indicator_check=_get_disable_flags(ctx)[1],
)
if list_modules:
@@ -295,8 +266,6 @@ def check_fs(ctx, iocs, output, fast, list_modules, module, hashes, verbose, dum
module_name=module,
module_options=module_options,
hashes=hashes,
disable_version_check=_get_disable_flags(ctx)[0],
disable_indicator_check=_get_disable_flags(ctx)[1],
)
if list_modules:
@@ -331,13 +300,7 @@ def check_fs(ctx, iocs, output, fast, list_modules, module, hashes, verbose, dum
@click.argument("FOLDER", type=click.Path(exists=True))
@click.pass_context
def check_iocs(ctx, iocs, list_modules, module, folder):
cmd = CmdCheckIOCS(
target_path=folder,
ioc_files=iocs,
module_name=module,
disable_version_check=_get_disable_flags(ctx)[0],
disable_indicator_check=_get_disable_flags(ctx)[1],
)
cmd = CmdCheckIOCS(target_path=folder, ioc_files=iocs, module_name=module)
cmd.modules = BACKUP_MODULES + FS_MODULES + MIXED_MODULES
if list_modules:

View File

@@ -24,8 +24,6 @@ class CmdIOSCheckBackup(Command):
serial: Optional[str] = None,
module_options: Optional[dict] = None,
hashes: bool = False,
disable_version_check: bool = False,
disable_indicator_check: bool = False,
) -> None:
super().__init__(
target_path=target_path,
@@ -36,8 +34,6 @@ class CmdIOSCheckBackup(Command):
module_options=module_options,
hashes=hashes,
log=log,
disable_version_check=disable_version_check,
disable_indicator_check=disable_indicator_check,
)
self.name = "check-backup"

View File

@@ -24,19 +24,16 @@ class CmdIOSCheckFS(Command):
serial: Optional[str] = None,
module_options: Optional[dict] = None,
hashes: bool = False,
disable_version_check: bool = False,
disable_indicator_check: bool = False,
) -> None:
super().__init__(
target_path=target_path,
results_path=results_path,
ioc_files=ioc_files,
module_name=module_name,
serial=serial,
module_options=module_options,
hashes=hashes,
log=log,
disable_version_check=disable_version_check,
disable_indicator_check=disable_indicator_check,
)
self.name = "check-fs"

View File

@@ -891,14 +891,6 @@
"version": "15.8.2",
"build": "19H384"
},
{
"version": "15.8.4",
"build": "19H390"
},
{
"version": "15.8.5",
"build": "19H394"
},
{
"build": "20A362",
"version": "16.0"
@@ -1000,14 +992,6 @@
"version": "16.7.8",
"build": "20H343"
},
{
"version": "16.7.11",
"build": "20H360"
},
{
"version": "16.7.12",
"build": "20H364"
},
{
"version": "17.0",
"build": "21A327"
@@ -1092,10 +1076,6 @@
"version": "17.6.1",
"build": "21G101"
},
{
"version": "17.7.7",
"build": "21H433"
},
{
"version": "18",
"build": "22A3354"
@@ -1127,41 +1107,5 @@
{
"version": "18.3.1",
"build": "22D72"
},
{
"version": "18.4",
"build": "22E240"
},
{
"version": "18.4.1",
"build": "22E252"
},
{
"version": "18.5",
"build": "22F76"
},
{
"version": "18.6",
"build": "22G86"
},
{
"version": "18.6.1",
"build": "22G90"
},
{
"version": "18.6.2",
"build": "22G100"
},
{
"version": "18.7",
"build": "22H20"
},
{
"version": "26",
"build": "23A341"
},
{
"version": "26.0.1",
"build": "23A355"
}
]

View File

@@ -43,8 +43,6 @@ class GlobalPreferences(IOSExtraction):
self.log.warning("Lockdown mode enabled")
else:
self.log.warning("Lockdown mode disabled")
return
self.log.warning("Lockdown mode disabled")
def process_file(self, file_path: str) -> None:
with open(file_path, "rb") as handle:

Some files were not shown because too many files have changed in this diff Show More