Compare commits

...
Author SHA1 Message Date
Donncha Ó Cearbhaill 03012d2c2d Fix CI coverage comment and test each matrix Python version
The coverage comment never posted: the per-file table with a link on
every file and missing line range exceeds GitHub's 65536-character
comment limit, so PRs got only a badge. Every matrix job also raced to
post the same comment, and the action was unpinned at @main.

Post from one job only, limit the table to files changed in the PR,
drop per-line links, and pin the action. Write the full coverage table
to the job summary as well, which also works for PRs from forks where
the token is read-only.

Set UV_PYTHON from the matrix. Without it, .python-version pins 3.10
and `uv run` rebuilt the venv with 3.10 after `uv sync --python X`, so
all five matrix jobs were testing on Python 3.10.
2026-09-04 20:25:07 +02:00
besendorf a463e8509c Merge pull request #774 from mvt-project/backup_extraction
Replace iOSbackup with iphone_backup_decrypt
2026-09-02 19:29:12 +02:00
Janik Besendorf ca063a2ddc Merge current main into backup decryption 2026-09-02 19:27:52 +02:00
Donncha Ó Cearbhaill ece44a7e7f Merge pull request #910 from mvt-project/feature/monthly-releases
Release monthly from a scheduled workflow
2026-09-01 19:41:01 +02:00
Donncha Ó Cearbhaill 27cfe7f1e7 Release weekly instead of monthly 2026-09-01 19:31:23 +02:00
Donncha Ó Cearbhaill 531d63ab07 Release monthly from a scheduled workflow
The package version now comes from the latest v* git tag, so a release
is a tag and nothing else. A scheduled workflow tags main on the first
of the month when something was merged since the last release, creates
the GitHub release with generated notes, and publishes to PyPI and the
container registry. Pushing a v* tag by hand goes through the same path.
2026-09-01 15:49:25 +02:00
Donncha Ó Cearbhaill b18afb4a48 Merge pull request #871 from mvt-project/fix/bugreport-parser-coverage
Fix Android bugreport parser coverage
2026-08-30 14:50:39 +02:00
Donncha Ó Cearbhaill 32c88822a0 Merge branch 'main' into fix/bugreport-parser-coverage 2026-08-30 14:11:54 +02:00
Donncha Ó Cearbhaill 23565af4c1 Merge pull request #907 from mvt-project/feature/fast-cli-startup
Keep the start of mvt-ios and mvt-android cheap for shell completion
2026-08-28 01:03:56 +02:00
Donncha Ó Cearbhaill 24c65859ee Merge branch 'main' into fix/bugreport-parser-coverage 2026-08-27 21:06:30 +02:00
Janik Besendorf 65df483258 Parse multiline Android properties 2026-08-25 19:19:47 +02:00
Janik Besendorf fd27fe3378 Fix module registry type annotations 2026-08-22 14:21:51 +02:00
Janik Besendorf 7ef26779bf Register added bugreport parsers 2026-08-22 14:21:51 +02:00
Janik Besendorf 3a43f6fcc7 Merge plaintext and protobuf tombstones 2026-08-22 14:21:51 +02:00
Janik Besendorf 45fdf85790 Add bugreport mountinfo parser 2026-08-22 14:21:51 +02:00
Janik Besendorf 627ab32d42 Add bugreport settings parser 2026-08-22 14:21:51 +02:00
Janik Besendorf 8b85972cc3 Add bugreport process table parser 2026-08-22 14:21:51 +02:00
Janik Besendorf f747751b09 Add timezone provenance to file timestamps 2026-08-22 14:21:51 +02:00
Janik Besendorf 832e46604d Parse all platform compatibility overrides 2026-08-22 14:21:51 +02:00
Janik Besendorf a3c8b56102 Parse typed multi-user package details 2026-08-22 14:21:51 +02:00
Janik Besendorf e9bf197ec6 Parse all database operation details 2026-08-22 14:20:59 +02:00
Janik Besendorf 288b313649 Parse complete battery history events 2026-08-22 14:20:59 +02:00
Janik Besendorf 87a27f6e37 Normalize battery daily update records 2026-08-22 14:20:59 +02:00
Janik Besendorf 1bf2f9b35a Retain AppOps UID and event details 2026-08-22 14:20:59 +02:00
Janik Besendorf b904414bb0 Parse ADB binary XML keys 2026-08-22 14:20:59 +02:00
Janik Besendorf 008844b440 Parse accessibility service states per user 2026-08-22 14:20:59 +02:00
Janik Besendorf ab879bb23b Parse all package resolver categories 2026-08-22 14:20:59 +02:00
Janik Besendorf dc0650bf76 Fix bugreport command section extraction 2026-08-22 14:20:59 +02:00
Janik Besendorf 545ac19158 Restore concurrent backup decryption 2026-08-17 13:24:30 +02:00
Codex 2de8312349 Merge main and fix encrypted backup extraction 2026-08-16 22:39:08 +05:30
besendorf 8b195d2696 Merge branch 'main' into backup_extraction 2026-04-27 15:03:42 +02:00
Janik Besendorf 39a26d0f0b Replace iOSbackup with iphone_backup_decrypt
Replace the unmaintained iOSbackup dependency with iphone_backup_decrypt
(MIT licensed, actively maintained). This fixes file corruption caused by
iOSbackup truncating files to inaccurate sizes from backup metadata.

The extract-key command and --key-file option are preserved via an
MVTEncryptedBackup subclass that patches the keybag unlock to
capture/reuse the derived PBKDF2 key.

Closes #669
2026-04-12 10:51:56 +02:00
63 changed files with 2074 additions and 889 deletions

No files matched your search

@@ -37,6 +37,8 @@ jobs:
steps:
- name: Checkout repository
uses: actions/checkout@v7
with:
fetch-depth: 0 # the package version comes from the git tags
# Uses the `docker/login-action` action to log in to the Container registry registry using the account and password that will publish the packages. Once published, the packages are scoped to the account defined here.
- name: Log in to the Container registry
uses: docker/login-action@dbcb813823bdd20940b903addbd779551569679f # v4.6.0
+44
View File
@@ -0,0 +1,44 @@
name: Release
on:
schedule:
- cron: "0 9 * * 1" # Mondays, 09:00 UTC
workflow_dispatch:
push:
tags: ["v*"]
jobs:
tag:
if: github.ref_type != 'tag'
runs-on: ubuntu-latest
permissions:
contents: write
actions: write
env:
GH_TOKEN: ${{ github.token }}
steps:
- uses: actions/checkout@v7
with:
fetch-depth: 0
- run: |
last=$(git describe --tags --abbrev=0 --match 'v*')
if [ "$(git rev-list "$last..HEAD" --count)" = 0 ]; then
echo "Nothing merged since $last"; exit 0
fi
tag=v$(date -u +%Y.%-m.%-d)
gh release create "$tag" --target "$GITHUB_SHA" --generate-notes
# Events made with GITHUB_TOKEN don't start other workflows; run them by hand.
gh workflow run release.yml --ref "$tag"
gh workflow run publish-release-docker.yml --ref "$tag"
publish:
if: github.ref_type == 'tag'
runs-on: ubuntu-latest
permissions:
id-token: write # PyPI trusted publishing
steps:
- uses: actions/checkout@v7
with:
fetch-depth: 0
- uses: astral-sh/setup-uv@v10.0.0
- run: uv build && uv publish
+19 -4
View File
@@ -5,6 +5,10 @@ on:
pull_request:
branches: [ main ]
permissions:
contents: read
pull-requests: write # coverage comment
jobs:
build:
name: Run Python Tests
@@ -13,6 +17,10 @@ jobs:
fail-fast: false
matrix:
python-version: ['3.10', '3.11', '3.12', '3.13', '3.14']
env:
# Takes precedence over .python-version, which otherwise makes `uv run`
# rebuild the venv with 3.10 and test every matrix entry on 3.10.
UV_PYTHON: ${{ matrix.python-version }}
steps:
- uses: actions/checkout@v7
@@ -26,16 +34,23 @@ jobs:
python-version: ${{ matrix.python-version }}
- name: Install Python dependencies
run: |
uv sync --locked --group dev --python ${{ matrix.python-version }}
uv sync --locked --group dev
- name: Test with pytest
run: |
set -o pipefail
make test-ci | tee pytest-coverage.txt
- name: Coverage job summary
run: uv run coverage report --format=markdown --show-missing --skip-covered >> "$GITHUB_STEP_SUMMARY"
- name: Pytest coverage comment
continue-on-error: true # Workflows running on a fork can't post comments
uses: MishaKav/pytest-coverage-comment@main
if: github.event_name == 'pull_request'
# One comment per PR, not one per matrix entry. PRs from forks get a
# read-only token and can't post; the job summary above still works.
if: github.event_name == 'pull_request' && matrix.python-version == '3.13'
continue-on-error: true
uses: MishaKav/pytest-coverage-comment@v1.12.2
with:
pytest-coverage-path: ./pytest-coverage.txt
junitxml-path: ./pytest.xml
# The full table with per-line links exceeds GitHub's 65536-char comment limit.
report-only-changed-files: true
remove-links-to-lines: true
+1
View File
@@ -122,6 +122,7 @@ RUN apt-get update \
binutils \
default-jre-headless \
file \
git \
jq \
less \
libcurl4 \
-5
View File
@@ -32,8 +32,3 @@ clean:
dist:
$(UV) build
upload:
$(UV) tool run twine upload dist/*
test-upload:
$(UV) tool run twine upload --repository testpypi dist/*
+6 -4
View File
@@ -24,7 +24,8 @@ dependencies = [
"simplejson==4.1.1",
"packaging==26.3",
"appdirs==1.4.4",
"iOSbackup==0.9.925",
"iphone_backup_decrypt==0.9.0",
"pycryptodome>=3.20.0",
"adb-shell[usb]==0.4.4",
"libusb1==3.4.0",
"cryptography==50.0.0",
@@ -72,7 +73,7 @@ docs = [
]
[build-system]
requires = ["setuptools>=61.0"]
requires = ["setuptools>=61.0", "setuptools-scm>=8"]
build-backend = "setuptools.build_meta"
[tool.coverage.run]
@@ -121,5 +122,6 @@ where = ["src"]
[tool.setuptools.package-data]
mvt = ["ios/data/*.json"]
[tool.setuptools.dynamic]
version = { attr = "mvt.common.version.MVT_VERSION" }
[tool.setuptools_scm]
# The version is the latest v* tag; ignore the archive/* tags.
git_describe_command = "git describe --dirty --tags --long --match 'v*'"
@@ -4,6 +4,7 @@
# https://license.mvt.re/1.1/
import re
from typing import Any
from .artifact import AndroidArtifact
@@ -20,10 +21,10 @@ class DumpsysAccessibilityArtifact(AndroidArtifact):
continue
self.alertstore.medium(
f'Found accessibility service: "{result["service"]}"',
f'Found accessibility service: "{result["component"]}"',
"",
result,
)
)
def parse(self, content: str) -> None:
"""
@@ -33,41 +34,69 @@ class DumpsysAccessibilityArtifact(AndroidArtifact):
:param content: content of the accessibility section (string)
"""
# "Old" syntax
in_services = False
self.results: list[dict[str, Any]] = []
services: dict[tuple[int | None, str], dict] = {}
user_id: int | None = None
state: str | None = None
for line in content.splitlines():
if line.strip().startswith("installed services:"):
in_services = True
continue
user_match = re.search(r"attributes:\{id=(\d+)", line)
if user_match:
user_id = int(user_match.group(1))
if not in_services:
continue
if line.strip() == "}":
# At end of installed services
break
service = line.split(":")[1].strip()
self.results.append(
{
"package_name": service.split("/")[0],
"service": service,
}
stripped = line.strip()
state_match = re.match(
r"(?i)(installed|enabled|binding|bound|crashed) services\s*:\s*\{(.*)",
stripped,
)
# "New" syntax - AOSP >= 14 (?)
# Looks like:
# Enabled services:{{com.azure.authenticator/com.microsoft.brooklyn.module.accessibility.BrooklynAccessibilityService}, {com.agilebits.onepassword/com.agilebits.onepassword.filling.accessibility.FillingAccessibilityService}}
for line in content.splitlines():
if line.strip().startswith("Enabled services:"):
matches = re.finditer(r"{([^{]+?)}", line)
for match in matches:
# Each match is in format: <package_name>/<service>
package_name, _, service = match.group(1).partition("/")
self.results.append(
{"package_name": package_name, "service": service}
if state_match:
state = state_match.group(1).lower()
inline = state_match.group(2)
for component in re.findall(
r"\{?([\w.$-]+/[\w.$-]+)(?:\s+\(A11yTool\))?\}?", inline
):
service = services.setdefault(
(user_id, component), self._new_service(component, user_id)
)
service[self._state_field(state)] = True
service["accessibility_tool"] = "(A11yTool)" in inline
continue
if not state:
continue
if stripped == "}" or stripped.startswith("AccessibilityInputFilter"):
state = None
continue
component_match = re.search(
r"(?:\d+\s*:\s*)?([\w.$-]+/[\w.$-]+)(?:\s+\(A11yTool\))?",
stripped,
)
if component_match:
component = component_match.group(1)
service = services.setdefault(
(user_id, component), self._new_service(component, user_id)
)
service[self._state_field(state)] = True
service["accessibility_tool"] = "(A11yTool)" in stripped
self.results.extend(services.values())
@staticmethod
def _state_field(state: str) -> str:
return {"binding": "binding", "bound": "bound"}.get(state, state)
@staticmethod
def _new_service(component: str, user_id: int | None) -> dict:
package_name, service_name = component.split("/", 1)
return {
"user_id": user_id,
"component": component,
"package_name": package_name,
"service_name": service_name,
"installed": False,
"enabled": False,
"binding": False,
"bound": False,
"crashed": False,
"accessibility_tool": False,
}
+41 -1
View File
@@ -6,6 +6,7 @@
import base64
import binascii
import hashlib
import re
from .artifact import AndroidArtifact
@@ -98,6 +99,34 @@ class DumpsysADBArtifact(AndroidArtifact):
return keystore
def parse_binary_xml(self, data: bytes) -> list[dict]:
"""Recover ADB key records from Android binary XML (ABX).
Some dumpstate implementations embed ABX in a text stream and replace
binary token bytes. The public key remains intact, while unavailable
numeric metadata is represented as ``None`` rather than corrupt text.
"""
keystore = []
seen = set()
for match in re.finditer(
rb"(?<![A-Za-z0-9+/])([A-Za-z0-9+/]{300,}={0,2})"
rb"(?: ([A-Za-z0-9_.@-]+))?",
data,
):
key = match.group(1)
try:
base64.b64decode(key, validate=True)
except (binascii.Error, ValueError):
continue
if key in seen:
continue
seen.add(key)
full_key = key + (b" " + match.group(2) if match.group(2) else b"")
key_info = self.calculate_key_info(full_key)
key_info["last_connected"] = None
keystore.append(key_info)
return keystore
@staticmethod
def calculate_key_info(user_key: bytes) -> dict:
if b" " in user_key:
@@ -118,7 +147,7 @@ class DumpsysADBArtifact(AndroidArtifact):
return {
"user": user.decode("utf-8"),
"fingerprint": key_fingerprint_colon,
"key": key_base64,
"key": key_base64.decode("ascii", errors="replace"),
}
def check_indicators(self) -> None:
@@ -175,12 +204,23 @@ class DumpsysADBArtifact(AndroidArtifact):
# Keystore is in XML format on some devices and we need to parse it
if keystore_data and keystore_data.startswith(b"<?xml"):
parsed["debugging_manager"]["keystore"] = self.parse_xml(keystore_data)
elif keystore_data and keystore_data.startswith(b"ABX\x00"):
parsed["debugging_manager"]["keystore"] = self.parse_binary_xml(
keystore_data
)
else:
# Keystore is not XML format
parsed["debugging_manager"]["keystore"] = keystore_data
parsed = parsed["debugging_manager"]
for key, value in list(parsed.items()):
if isinstance(value, bytes):
decoded = value.decode("utf-8", errors="replace")
parsed[key] = (
decoded == "true" if decoded in ("true", "false") else decoded
)
# Calculate key fingerprints for better readability
key_info = []
for user_key in parsed.get("user_keys", []):
+122 -81
View File
@@ -4,6 +4,7 @@
# https://license.mvt.re/1.1/
from datetime import datetime
import re
from typing import Any
from mvt.common.module_types import ModuleAtomicResult, ModuleSerializedResult
@@ -27,14 +28,14 @@ class DumpsysAppopsArtifact(AndroidArtifact):
continue
for entry in perm["entries"]:
if "timestamp" in entry:
if entry.get("timestamp"):
records.append(
{
"timestamp": entry["timestamp"],
"module": self.__class__.__name__,
"event": entry["access"],
"event": entry["event"],
"data": f"{result['package_name']} access to "
f"{perm['name']}: {entry['access']}",
f"{perm['name']}: {entry['event']}",
}
)
@@ -51,7 +52,7 @@ class DumpsysAppopsArtifact(AndroidArtifact):
continue
# We use a placeholder entry to create a basic alert even without permission entries.
placeholder_entry = {"access": "Unknown", "timestamp": ""}
placeholder_entry = {"event": "unknown", "timestamp": ""}
for perm in result["permissions"]:
if (
@@ -60,12 +61,12 @@ class DumpsysAppopsArtifact(AndroidArtifact):
):
for entry in sorted(
perm["entries"] or [placeholder_entry],
key=lambda x: x["timestamp"],
key=lambda x: x.get("timestamp") or "",
):
cleaned_result = result.copy()
cleaned_result["permissions"] = [perm]
self.alertstore.medium(
f"Package '{result['package_name']}' had risky permission '{perm['name']}' set to '{entry['access']}' at {entry['timestamp']}",
f"Package '{result['package_name']}' had risky permission '{perm['name']}' set to '{entry['event']}' at {entry['timestamp']}",
entry["timestamp"],
cleaned_result,
)
@@ -73,111 +74,151 @@ class DumpsysAppopsArtifact(AndroidArtifact):
elif result["package_name"] in RISKY_PACKAGES:
for entry in sorted(
perm["entries"] or [placeholder_entry],
key=lambda x: x["timestamp"],
key=lambda x: x.get("timestamp") or "",
):
cleaned_result = result.copy()
cleaned_result["permissions"] = [perm]
self.alertstore.medium(
f"Risky package '{result['package_name']}' had '{perm['name']}' permission set to '{entry['access']}' at {entry['timestamp']}",
f"Risky package '{result['package_name']}' had '{perm['name']}' permission set to '{entry['event']}' at {entry['timestamp']}",
entry["timestamp"],
cleaned_result,
)
def parse(self, output: str) -> None:
# self.results: List[Dict[str, Any]] = []
perm: dict[str, Any] = {}
package: dict[str, Any] = {}
entry: dict[str, Any] = {}
uid = None
self.results: list[dict[str, Any]] = []
permission: dict[str, Any] | None = None
package: dict[str, Any] | None = None
uid: str | None = None
uid_details: dict[str, Any] = {}
attribution: str | None = None
in_packages = False
def finish_permission() -> None:
nonlocal permission
if package is not None and permission is not None:
package["permissions"].append(permission)
permission = None
def finish_package() -> None:
nonlocal package
finish_permission()
if package is not None:
self.results.append(package)
package = None
for line in output.splitlines():
if line.startswith(" Uid 0:"):
uid_match = re.match(r"^ Uid ([^:]+):$", line)
if uid_match:
in_packages = True
finish_package()
uid = uid_match.group(1)
uid_details = {
"uid_state": None,
"capability": None,
"app_widget_visible": None,
"default_modes": {},
}
continue
if not in_packages:
continue
if line.startswith(" Uid "):
uid = line[6:-1]
if entry:
perm["entries"].append(entry)
entry = {}
if package:
if perm:
package["permissions"].append(perm)
uid_property = re.match(
r"^ (state|capability|appWidgetVisible)=(.*)$", line
)
if uid_property:
key = {
"state": "uid_state",
"appWidgetVisible": "app_widget_visible",
}.get(uid_property.group(1), uid_property.group(1))
value: Any = uid_property.group(2)
if value in ("true", "false"):
value = value == "true"
uid_details[key] = value
continue
perm = {}
self.results.append(package)
package = {}
default_mode = re.match(r"^ ([A-Z0-9_]+): mode=([^\s]+)", line)
if default_mode and package is None:
uid_details["default_modes"][default_mode.group(1)] = (
default_mode.group(2)
)
continue
if line.startswith(" Package "):
if entry:
perm["entries"].append(entry)
entry = {}
if package:
if perm:
package["permissions"].append(perm)
perm = {}
self.results.append(package)
finish_package()
package = {
"package_name": line[12:-1],
"permissions": [],
"uid": uid,
**uid_details,
}
continue
if package and line.startswith(" ") and line[6] != " ":
if entry:
perm["entries"].append(entry)
entry = {}
if perm:
package["permissions"].append(perm)
perm = {}
perm["name"] = line.split()[0]
perm["entries"] = []
if len(line.split()) > 1:
perm["access"] = line.split()[1][1:-2]
operation_match = re.match(
r"^ ([A-Z0-9_]+)(?: \(([^)]+)\))?:\s*$", line
)
if package is not None and operation_match:
finish_permission()
permission = {
"name": operation_match.group(1),
"mode": operation_match.group(2),
"entries": [],
}
attribution = None
continue
if line.startswith(" "):
# Permission entry like:
# Reject: [fg-s]2021-05-19 22:02:52.054 (-314d1h25m2s33ms)
access_type = line.split(":")[0].strip()
if access_type not in ["Access", "Reject"]:
# Skipping invalid access type. Some entries are not in the format we expect
continue
attribution_match = re.match(r"^\s{8,}([^=]+)=\[$", line)
if attribution_match:
attribution = attribution_match.group(1).strip()
continue
if line.strip() == "]":
attribution = None
continue
if entry:
perm["entries"].append(entry)
entry = {}
entry["access"] = access_type
entry["type"] = line[line.find("[") + 1 : line.find("]")]
try:
entry["timestamp"] = convert_datetime_to_iso(
datetime.strptime(
line[line.find("]") + 1 : line.find("(")].strip(),
"%Y-%m-%d %H:%M:%S.%f",
)
if permission is None:
continue
event_match = re.match(
r"^\s*(Access|Reject):\s*\[([^]]+)\]\s*"
r"(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d+)\s*"
r"(\([^)]*\))?(?:\s+duration=([^\s]+))?",
line,
)
running_match = re.match(
r"^\s*Running start at:\s*(\S+(?: \S+)?)",
line,
)
if event_match:
entry = {
"event": event_match.group(1).lower(),
"access": event_match.group(1),
"uid_state": event_match.group(2),
"timestamp": convert_datetime_to_iso(
datetime.strptime(event_match.group(3), "%Y-%m-%d %H:%M:%S.%f")
),
"relative_time": event_match.group(4),
"duration": event_match.group(5),
"attribution": attribution,
}
permission["entries"].append(entry)
elif running_match:
raw_start = running_match.group(1)
timestamp = None
if re.fullmatch(r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d+", raw_start):
timestamp = convert_datetime_to_iso(
datetime.strptime(raw_start, "%Y-%m-%d %H:%M:%S.%f")
)
except ValueError:
# Invalid date format
pass
permission["entries"].append(
{
"event": "running",
"access": "Running",
"uid_state": None,
"timestamp": timestamp,
"relative_time": raw_start
if raw_start.startswith("+")
else None,
"duration": None,
"attribution": attribution,
}
)
if line.strip() == "":
break
if entry:
perm["entries"].append(entry)
if perm:
package["permissions"].append(perm)
if package:
self.results.append(package)
finish_package()
@@ -18,18 +18,18 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact):
def serialize(self, record: ModuleAtomicResult) -> ModuleSerializedResult:
action = record.get("action", "update")
package_name = record["package_name"]
vers = record["vers"]
vers = record["version_code"]
if vers == "0":
if vers == 0:
data = f"Recorded uninstall of package {package_name} (vers 0)"
elif action == "downgrade":
prev_vers = record.get("previous_vers", "unknown")
prev_vers = record.get("previous_version_code", "unknown")
data = f"Recorded downgrade of package {package_name} from vers {prev_vers} to vers {vers}"
else:
data = f"Recorded update of package {package_name} with vers {vers}"
return {
"timestamp": record["from"],
"timestamp": record["period_start"],
"module": self.__class__.__name__,
"event": "battery_daily",
"data": data,
@@ -48,6 +48,7 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact):
continue
def parse(self, output: str) -> None:
self.results = []
daily = None
daily_updates: list[dict[str, Any]] = []
records: list[dict[str, Any]] = []
@@ -59,7 +60,10 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact):
timeframe = line[13:].strip()
date_from, date_to = timeframe.strip(":").split(" to ", 1)
daily = {"from": date_from[0:10], "to": date_to[0:10]}
daily = {
"period_start": self._format_daily_timestamp(date_from),
"period_end": self._format_daily_timestamp(date_to),
}
continue
if not daily:
@@ -70,21 +74,30 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact):
line = line.strip().replace("Update ", "")
package_name, vers = line.split(" ", 1)
vers_nr = vers.split("=", 1)[1]
vers_raw = vers.split("=", 1)[1]
try:
version_code: int | str = int(vers_raw)
except ValueError:
version_code = vers_raw
already_seen = False
for update in daily_updates:
if package_name == update["package_name"] and vers_nr == update["vers"]:
if (
package_name == update["package_name"]
and version_code == update["version_code"]
):
update["occurrences"] += 1
already_seen = True
break
if not already_seen:
update_record: dict[str, Any] = {
"action": "update",
"from": daily["from"],
"to": daily["to"],
"period_start": daily["period_start"],
"period_end": daily["period_end"],
"package_name": package_name,
"vers": vers_nr,
"version_code": version_code,
"occurrences": 1,
}
daily_updates.append(update_record)
@@ -95,26 +108,31 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact):
self._detect_uninstalls_and_downgrades(records)
self.results.extend(records)
def _detect_uninstalls_and_downgrades(
self, records: list[dict[str, Any]]
) -> None:
@staticmethod
def _format_daily_timestamp(value: str) -> str:
if len(value) >= 19 and value[10] == "-":
return f"{value[:10]} {value[11:].replace('-', ':')}"
return value
def _detect_uninstalls_and_downgrades(self, records: list[dict[str, Any]]) -> None:
package_versions: dict[str, int] = {}
for record in sorted(
records,
key=lambda record: (
record["from"],
record["to"],
record["period_start"],
record["period_end"],
record["package_name"],
),
):
package_name = record["package_name"]
vers_nr = record["vers"]
vers_nr = record["version_code"]
if vers_nr == "0":
if vers_nr == 0:
record["action"] = "uninstall"
self.alertstore.medium(
f"Detected uninstall of package {package_name} (vers 0)",
record["from"],
record["period_start"],
record,
)
package_versions.pop(package_name, None)
@@ -128,11 +146,11 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact):
previous_vers = package_versions.get(package_name)
if previous_vers is not None and current_vers < previous_vers:
record["action"] = "downgrade"
record["previous_vers"] = str(previous_vers)
record["previous_version_code"] = previous_vers
self.alertstore.medium(
f"Detected downgrade of package {package_name} "
f"from vers {previous_vers} to vers {current_vers}",
record["from"],
record["period_start"],
record,
)
@@ -1,96 +1,177 @@
# Mobile Verification Toolkit (MVT)
# Copyright (c) 2021-2023 The MVT Authors.
# 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 datetime
import re
from mvt.common.utils import convert_datetime_to_iso
from .artifact import AndroidArtifact
class DumpsysBatteryHistoryArtifact(AndroidArtifact):
"""
Parser for dumpsys dattery history events.
"""
"""Parser for package-related events in dumpsys batterystats history."""
def check_indicators(self) -> None:
if not self.indicators:
return
for result in self.results:
ioc_match = self.indicators.check_app_id(result["package_name"])
package_name = result.get("package_name")
if not package_name:
continue
ioc_match = self.indicators.check_app_id(package_name)
if ioc_match:
self.alertstore.critical(
ioc_match.message, "", result, matched_indicator=ioc_match.ioc
)
continue
@staticmethod
def _parse_wall_time(value: str) -> datetime.datetime | None:
if re.fullmatch(r"\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d+", value):
value = f"1900-{value}"
for date_format in (
"%Y-%m-%d-%H-%M-%S-%f",
"%Y-%m-%d-%H-%M-%S",
"%Y-%m-%d %H:%M:%S.%f",
):
try:
return datetime.datetime.strptime(value, date_format)
except ValueError:
pass
return None
@staticmethod
def _elapsed_seconds(value: str) -> float | None:
if not value.startswith("+"):
return None
units = {"d": 86400, "h": 3600, "m": 60, "s": 1, "ms": 0.001}
total = 0.0
for number, unit in re.findall(r"(\d+)(ms|d|h|m|s)", value):
total += int(number) * units[unit]
return total
@staticmethod
def _package_from_name(name: str) -> str | None:
clean = name.removeprefix("*walarm*:").removeprefix("*alarm*:")
slash_parts = clean.split("/")
if len(slash_parts) > 1:
first = slash_parts[0].lstrip("@")
if first.startswith(("com.", "org.", "net.")):
return first
for part in reversed(slash_parts[1:]):
candidate = part.lstrip("@").split(":", 1)[0]
if candidate.startswith(("com.", "org.", "net.")):
return candidate
parts = clean.split(".")
package_parts = []
for part in parts:
if part and part[0].islower():
package_parts.append(part)
else:
break
return ".".join(package_parts) if len(package_parts) >= 2 else None
@staticmethod
def _normalize_service(service: str) -> str:
# WorkManager decorates jobs with one or more scheduler prefixes.
if "@" in service:
candidates = [part for part in service.split("@") if "/" in part]
if candidates:
return candidates[-1]
return service
def parse(self, data: str) -> None:
self.results: list[dict[str, str | None]] = []
anchor_time: datetime.datetime | None = None
anchor_elapsed = 0.0
has_history_heading = any(
line.startswith("Battery History") for line in data.splitlines()
)
in_history = not has_history_heading
for line in data.splitlines():
if line.startswith("Battery History "):
stripped = line.strip()
if line.startswith("Battery History"):
if in_history:
break
in_history = True
continue
if line.strip() == "":
if not in_history:
continue
if has_history_heading and not stripped:
break
time_parts = line.strip().split()
time_elapsed = time_parts[0]
if (
len(time_parts) > 1
and len(time_parts[0]) == 5
and time_parts[0][2] == "-"
and ":" in time_parts[1]
):
time_elapsed = " ".join(time_parts[:2])
event = ""
if line.find("+job") > 0:
event = "start_job"
payload = line.split("+job=", 1)[1]
uid, separator, service = payload.partition(":")
if not separator:
continue
service = service.strip().strip('"')
package_name = service.split("/")[0]
elif line.find("-job") > 0:
event = "end_job"
payload = line.split("-job=", 1)[1]
uid, separator, service = payload.partition(":")
if not separator:
continue
service = service.strip().strip('"')
package_name = service.split("/")[0]
elif line.find("+running +wake_lock=") > 0:
payload = line.split("+running +wake_lock=", 1)[1]
uid, separator, _ = payload.partition(":")
if not separator:
continue
event = "wake"
service = (
line[line.find("*walarm*:") + 9 :].split(" ")[0].strip('"').strip()
)
if service == "" or "/" not in service:
continue
package_name = service.split("/")[0]
elif (line.find("+top=") > 0) or (line.find("-top") > 0):
if line.find("+top=") > 0:
event = "start_top"
top_pos = line.find("+top=")
else:
event = "end_top"
top_pos = line.find("-top=")
colon_pos = top_pos + line[top_pos:].find(":")
uid = line[top_pos + 5 : colon_pos]
service = ""
package_name = line[colon_pos + 1 :].strip('"')
else:
reset_match = re.search(r"(?:RESET:)?TIME:\s*(\S+)", stripped)
if reset_match:
parsed_time = self._parse_wall_time(reset_match.group(1))
if parsed_time is not None:
elapsed_token = stripped.split()[0]
anchor_elapsed = self._elapsed_seconds(elapsed_token) or 0.0
anchor_time = parsed_time
continue
self.results.append(
{
"time_elapsed": time_elapsed,
"event": event,
"uid": uid,
"package_name": package_name,
"service": service,
}
)
fields = stripped.split()
if not fields:
continue
if len(fields) > 1 and re.fullmatch(r"\d{2}-\d{2}", fields[0]):
time_elapsed = " ".join(fields[:2])
line_time = self._parse_wall_time(time_elapsed)
elapsed = None
else:
time_elapsed = fields[0]
elapsed = self._elapsed_seconds(time_elapsed)
line_time = None
timestamp = line_time
if timestamp is None and anchor_time is not None and elapsed is not None:
timestamp = anchor_time + datetime.timedelta(
seconds=elapsed - anchor_elapsed
)
def add(
event: str, uid: str, service: str, package_name: str | None
) -> None:
self.results.append(
{
"time_elapsed": time_elapsed,
"timestamp": convert_datetime_to_iso(timestamp)
if timestamp
else None,
"event": event,
"uid": uid,
"package_name": package_name,
"service": service,
}
)
for sign, uid, raw_service in re.findall(
r"([+-])job=([^:\s]+):\"([^\"]+)\"", line
):
service = self._normalize_service(raw_service)
add(
"start_job" if sign == "+" else "end_job",
uid,
service,
self._package_from_name(service) or service.split("/", 1)[0],
)
for sign, uid, package_name in re.findall(
r"([+-])top=([^:\s]+):\"([^\"]+)\"", line
):
add(
"start_top" if sign == "+" else "end_top",
uid,
"",
package_name,
)
wake_match = re.search(r"\+wake_lock=([^:\s]+):\"([^\"]+)\"", line)
if wake_match:
wake_name = wake_match.group(2)
add(
"wake",
wake_match.group(1),
wake_name,
self._package_from_name(wake_name),
)
+25 -11
View File
@@ -29,16 +29,27 @@ class DumpsysDBInfoArtifact(AndroidArtifact):
def parse(self, output: str) -> None:
rxp = re.compile(
r".*\[((?:[0-9]{4}-)?[0-9]{2}-[0-9]{2} "
r"[0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3})\]\s*"
r"(?:\[Pid:\((\d+)\)\])?([\w-]+).*?sql=\"(.+?)\""
) # pylint: disable=line-too-long
r"^\s*\d+:\s*\[((?:\d{4}-)?\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{3})\]\s*"
r"(?:\[Pid:\((\d+)\)\])?\s*([\w-]+) took (\d+)ms - ([^,]+),"
r"\s*sql=\"(.*)\"(?:, path=(.*))?$"
)
pool = None
pool: str | None = None
connection_number: int | None = None
is_primary: bool | None = None
in_operations = False
for line in output.splitlines():
if line.startswith("Connection pool for "):
pool = line.replace("Connection pool for ", "").rstrip(":")
in_operations = False
connection_match = re.match(r"\s+Connection #(\d+):", line)
if connection_match:
connection_number = int(connection_match.group(1))
is_primary = None
if line.strip().startswith("isPrimaryConnection:"):
is_primary = line.strip().split(":", 1)[1].strip() == "true"
if not pool:
continue
@@ -52,7 +63,6 @@ class DumpsysDBInfoArtifact(AndroidArtifact):
if not line.startswith(" "):
in_operations = False
pool = None
continue
match = rxp.match(line)
@@ -60,11 +70,15 @@ class DumpsysDBInfoArtifact(AndroidArtifact):
continue
result = {
"isodate": match.group(1),
"timestamp": match.group(1),
"pid": int(match.group(2)) if match.group(2) else None,
"action": match.group(3),
"sql": match.group(4),
"path": pool,
"duration_ms": int(match.group(4)),
"status": match.group(5),
"sql": match.group(6),
"path": match.group(7) or pool,
"pool_path": pool,
"connection_number": connection_number,
"is_primary": is_primary,
}
if match.group(2):
result["pid"] = match.group(2)
self.results.append(result)
@@ -4,6 +4,7 @@
# https://license.mvt.re/1.1/
from .artifact import AndroidArtifact
from .package_resolvers import parse_resolver_table
class DumpsysPackageActivitiesArtifact(AndroidArtifact):
@@ -19,67 +20,11 @@ class DumpsysPackageActivitiesArtifact(AndroidArtifact):
)
continue
def parse(self, content: str):
def parse(self, content: str) -> None:
"""
Parse the Dumpsys Package section for activities
Adds results to self.results
:param content: content of the package section (string)
"""
self.results = []
in_activity_resolver_table = False
in_non_data_actions = False
intent = None
for line in content.splitlines():
if line.startswith("Activity Resolver Table:"):
in_activity_resolver_table = True
continue
if not in_activity_resolver_table:
continue
if line.startswith(" Non-Data Actions:"):
in_non_data_actions = True
continue
if not in_non_data_actions:
continue
# If we hit an empty line, the Non-Data Actions section should be
# finished.
if line.strip() == "":
break
# We detect the action name.
if (
line.startswith(" " * 6)
and not line.startswith(" " * 8)
and ":" in line
):
intent = line.strip().replace(":", "")
continue
# If we are not in an intent block yet, skip.
if not intent:
continue
# If we are in a block but the line does not start with 8 spaces
# it means the block ended a new one started, so we reset and
# continue.
if not line.startswith(" " * 8):
intent = None
continue
# If we got this far, we are processing receivers for the
# activities we are interested in.
activity = line.strip().split(" ")[1]
package_name = activity.split("/")[0]
self.results.append(
{
"intent": intent,
"package_name": package_name,
"activity": activity,
}
)
self.results = parse_resolver_table(content, "Activity")
+205 -183
View File
@@ -1,10 +1,10 @@
# Mobile Verification Toolkit (MVT)
# Copyright (c) 2021-2023 The MVT Authors.
# 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 re
from typing import Any, Dict, List, Optional
from typing import Any
from mvt.android.utils import ROOT_PACKAGES
from mvt.common.module_types import ModuleAtomicResult, ModuleSerializedResult
@@ -12,222 +12,244 @@ from mvt.common.module_types import ModuleAtomicResult, ModuleSerializedResult
from .artifact import AndroidArtifact
def _value(raw: str) -> Any:
if raw == "null":
return None
if raw in ("true", "false"):
return raw == "true"
try:
return int(raw)
except ValueError:
return raw
class DumpsysPackagesArtifact(AndroidArtifact):
def check_indicators(self) -> None:
alerted_root_packages = set()
for result in self.results:
if result["package_name"] in ROOT_PACKAGES:
if result["package_name"] in alerted_root_packages:
continue
alerted_root_packages.add(result["package_name"])
package_name = result["package_name"]
if (
package_name in ROOT_PACKAGES
and package_name not in alerted_root_packages
):
alerted_root_packages.add(package_name)
self.alertstore.medium(
f'Found an installed package related to rooting/jailbreaking: "{result["package_name"]}"',
f'Found an installed package related to rooting/jailbreaking: "{package_name}"',
"",
result,
)
continue
if not self.indicators:
continue
ioc_match = self.indicators.check_app_id(result.get("package_name", ""))
ioc_match = self.indicators.check_app_id(package_name)
if ioc_match:
self.alertstore.critical(
ioc_match.message, "", result, matched_indicator=ioc_match.ioc
)
def serialize(self, record: ModuleAtomicResult) -> ModuleSerializedResult:
records = []
timestamps = [
{"event": "package_install", "timestamp": record["timestamp"]},
("package_install", record.get("timestamp")),
("package_last_update", record.get("last_update_time")),
]
timestamps.extend(
("package_first_install", user.get("first_install_time"))
for user in record.get("users", [])
)
return [
{
"event": "package_first_install",
"timestamp": record["first_install_time"],
},
{"event": "package_last_update", "timestamp": record["last_update_time"]},
"timestamp": timestamp,
"module": self.__class__.__name__,
"event": event,
"data": f"Install or update of package {record['package_name']}",
}
for event, timestamp in timestamps
if timestamp
]
for timestamp in timestamps:
records.append(
{
"timestamp": timestamp["timestamp"],
"module": self.__class__.__name__,
"event": timestamp["event"],
"data": f"Install or update of package {record['package_name']}",
}
)
return records
@staticmethod
def parse_dumpsys_package_for_details(output: str) -> Dict[str, Any]:
"""
Parse one entry of a dumpsys package information
"""
details: Dict[str, Any] = {
"uid": "",
"version_name": "",
"version_code": "",
"timestamp": "",
"first_install_time": "",
"last_update_time": "",
"installer": "",
"system": False,
"permissions": list(),
"requested_permissions": list(),
def _permission(line: str, permission_type: str) -> dict:
name, _, details = line.strip().partition(":")
granted_match = re.search(r"granted=(true|false)", details)
flags_match = re.search(r"flags=\[\s*([^]]*)\]", details)
return {
"name": name,
"type": permission_type,
"granted": granted_match.group(1) == "true" if granted_match else None,
"flags": [
flag.strip()
for flag in (flags_match.group(1).split("|") if flags_match else [])
if flag.strip()
],
}
in_install_permissions = False
in_runtime_permissions = False
in_declared_permissions = False
in_requested_permissions = True
current_user: Optional[int] = None
first_install_times: Dict[Optional[int], str] = {}
runtime_permissions: Dict[Optional[int], List[Dict[str, Any]]] = {}
@classmethod
def parse_dumpsys_package_for_details(cls, output: str) -> dict[str, Any]:
details: dict[str, Any] = {
"app_id": None,
"version_name": None,
"version_code": None,
"min_sdk": None,
"target_sdk": None,
"timestamp": None,
"last_update_time": None,
"installer": None,
"system": False,
"permissions": [],
"requested_permissions": [],
"users": [],
}
permission_section: str | None = None
current_user: dict[str, Any] | None = None
legacy_first_install: str | None = None
for line in output.splitlines():
user_match = re.match(r"User (\d+):", line.strip())
stripped = line.strip()
user_match = re.match(r"User (\d+):\s*(.*)", stripped)
if user_match:
current_user = int(user_match.group(1))
current_user = {"user_id": int(user_match.group(1)), "permissions": []}
for key, raw in re.findall(r"(\w+)=([^\s]+)", user_match.group(2)):
clean_key = {
"notLaunched": "not_launched",
"installReason": "install_reason",
"uninstallReason": "uninstall_reason",
"dataDir": "data_dir",
}.get(key, re.sub(r"(?<!^)(?=[A-Z])", "_", key).lower())
current_user[clean_key] = _value(raw)
details["users"].append(current_user)
permission_section = None
continue
if in_install_permissions:
if line.startswith(" " * 4) and not line.startswith(" " * 6):
in_install_permissions = False
else:
lineinfo = line.strip().split(":")
permission = lineinfo[0]
granted = None
if "granted=" in lineinfo[1]:
granted = "granted=true" in lineinfo[1]
header = stripped.lower()
if header in {
"declared permissions:",
"install permissions:",
"requested permissions:",
"runtime permissions:",
}:
permission_section = header.split()[0]
continue
details["permissions"].append(
{"name": permission, "granted": granted, "type": "install"}
)
if in_runtime_permissions:
if not line.startswith(" " * 8):
in_runtime_permissions = False
else:
lineinfo = line.strip().split(":")
permission = lineinfo[0]
granted = None
if "granted=" in lineinfo[1]:
granted = "granted=true" in lineinfo[1]
runtime_permissions.setdefault(current_user, []).append(
{"name": permission, "granted": granted, "type": "runtime"}
)
if in_declared_permissions:
if not line.startswith(" " * 6):
in_declared_permissions = False
else:
permission = line.strip().split(":")[0]
details["permissions"].append(
{"name": permission, "type": "declared"}
)
if in_requested_permissions:
if not line.startswith(" " * 6):
in_requested_permissions = False
else:
details["requested_permissions"].append(line.strip())
if line.strip().startswith("userId="):
details["uid"] = line.split("=")[1].strip()
elif line.strip().startswith("versionName="):
details["version_name"] = line.split("=")[1].strip()
elif line.strip().startswith("versionCode="):
details["version_code"] = line.split("=", 1)[1].strip()
elif line.strip().startswith("timeStamp="):
details["timestamp"] = line.split("=")[1].strip()
elif line.strip().startswith("installerPackageName="):
details["installer"] = line.split("=", 1)[1].strip()
elif line.strip().startswith("pkgFlags="):
details["system"] = "SYSTEM" in line.split("=", 1)[1].split()
elif line.strip().startswith("firstInstallTime="):
first_install_times[current_user] = line.split("=", 1)[1].strip()
elif line.strip().startswith("lastUpdateTime="):
details["last_update_time"] = line.split("=")[1].strip()
elif line.strip() == "install permissions:":
in_install_permissions = True
elif line.strip() == "runtime permissions:":
in_runtime_permissions = True
elif line.strip() == "declared permissions:":
in_declared_permissions = True
elif line.strip() == "requested permissions:":
in_requested_permissions = True
if 0 in first_install_times:
details["first_install_time"] = first_install_times[0]
elif None in first_install_times:
details["first_install_time"] = first_install_times[None]
elif first_install_times:
details["first_install_time"] = next(iter(first_install_times.values()))
if 0 in runtime_permissions:
details["permissions"].extend(runtime_permissions[0])
elif None in runtime_permissions:
details["permissions"].extend(runtime_permissions[None])
elif runtime_permissions:
details["permissions"].extend(next(iter(runtime_permissions.values())))
return details
def parse_dumpsys_packages(self, output: str) -> List[Dict[str, Any]]:
"""
Parse the dumpsys package service data
"""
pkg_rxp = re.compile(r" Package \[(.+?)\].*")
results = []
package_name = None
package = {}
lines: list[str] = []
for line in output.splitlines():
if line.startswith(" Package ["):
if len(lines) > 0:
details = self.parse_dumpsys_package_for_details("\n".join(lines))
package.update(details)
results.append(package)
lines = []
package = {}
matches = pkg_rxp.findall(line)
if not matches:
if current_user is not None:
user_property = re.match(
r"(installReason|uninstallReason|dataDir|firstInstallTime)=(.*)",
stripped,
)
if user_property:
key = {
"installReason": "install_reason",
"uninstallReason": "uninstall_reason",
"dataDir": "data_dir",
"firstInstallTime": "first_install_time",
}[user_property.group(1)]
current_user[key] = _value(user_property.group(2))
continue
package_name = matches[0]
package["package_name"] = package_name
if permission_section == "requested" and line.startswith(" "):
details["requested_permissions"].append(stripped)
continue
if permission_section in ("declared", "install") and line.startswith(
" "
):
details["permissions"].append(
cls._permission(stripped, permission_section)
)
continue
if (
permission_section == "runtime"
and line.startswith(" ")
and current_user is not None
):
current_user["permissions"].append(cls._permission(stripped, "runtime"))
continue
if not package_name:
simple_match = re.match(
r"(appId|userId|versionName|timeStamp|lastUpdateTime|installerPackageName)=(.*)",
stripped,
)
if simple_match:
key = {
"appId": "app_id",
"userId": "app_id",
"versionName": "version_name",
"timeStamp": "timestamp",
"lastUpdateTime": "last_update_time",
"installerPackageName": "installer",
}[simple_match.group(1)]
raw_value = simple_match.group(2)
details[key] = (
_value(raw_value)
if key == "app_id" or raw_value == "null"
else raw_value
)
continue
if stripped.startswith("pkgFlags="):
details["system"] = "SYSTEM" in stripped.split("=", 1)[1].split()
continue
version_match = re.match(
r"versionCode=([^\s]+)(?:\s+minSdk=([^\s]+))?(?:\s+targetSdk=([^\s]+))?",
stripped,
)
if version_match:
details["version_code"] = _value(version_match.group(1))
details["min_sdk"] = (
_value(version_match.group(2)) if version_match.group(2) else None
)
details["target_sdk"] = (
_value(version_match.group(3)) if version_match.group(3) else None
)
elif stripped.startswith("firstInstallTime="):
legacy_first_install = stripped.split("=", 1)[1]
lines.append(line)
if legacy_first_install:
user_zero = next(
(user for user in details["users"] if user["user_id"] == 0), None
)
if user_zero is None:
user_zero = {"user_id": 0, "permissions": []}
details["users"].append(user_zero)
user_zero.setdefault("first_install_time", legacy_first_install)
return details
if len(lines) > 0:
details = self.parse_dumpsys_package_for_details("\n".join(lines))
package.update(details)
results.append(package)
def parse(self, content: str) -> None:
self.results: list[dict[str, Any]] = []
category: str | None = None
package: dict[str, Any] | None = None
block: list[str] = []
return results
def finish() -> None:
nonlocal package, block
if package is not None:
package.update(self.parse_dumpsys_package_for_details("\n".join(block)))
self.results.append(package)
package = None
block = []
def parse(self, content: str):
"""
Parse the Dumpsys Package section for activities
Adds results to self.results
:param content: content of the package section (string)
"""
self.results = []
package = []
in_package_list = False
for line in content.splitlines():
if line.startswith("Packages:"):
in_package_list = True
if line == "Packages:":
finish()
category = "active"
continue
if not in_package_list:
if line == "Hidden system packages:":
finish()
category = "hidden_system"
continue
if line.strip() == "":
break
package.append(line)
self.results = self.parse_dumpsys_packages("\n".join(package))
package_match = re.match(r"^ Package \[([^]]+)\]", line)
if package_match and category:
finish()
package = {
"package_name": package_match.group(1),
"package_type": category,
}
continue
if (
category
and line
and not line.startswith(" ")
and not line.endswith(" overlay paths:")
):
finish()
category = None
continue
if package is not None:
block.append(line)
finish()
@@ -3,6 +3,8 @@
# Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/
import re
from .artifact import AndroidArtifact
@@ -25,19 +27,37 @@ class DumpsysPlatformCompatArtifact(AndroidArtifact):
def parse(self, data: str) -> None:
for line in data.splitlines():
if not line.startswith("ChangeId(168419799; name=DOWNSCALED;"):
match = re.match(r"ChangeId\((\d+);\s*(.*)\)$", line.strip())
if not match or "rawOverrides={" not in line:
continue
if line.strip() == "":
break
# Look for rawOverrides field
if "rawOverrides={" in line:
# Extract the content inside the braces for rawOverrides
overrides_field = line.split("rawOverrides={", 1)[1].split("};", 1)[0]
for entry in overrides_field.split(", "):
# Extract app name
uninstall_app = entry.split("=")[0].strip()
self.results.append({"package_name": uninstall_app})
body = match.group(2)
name_match = re.search(r"(?:^|;\s*)name=([^;]+)", body)
state = (
"enabled"
if re.search(r"(?:^|;\s*)enabled(?:;|$)", body)
else "disabled"
)
overridable = bool(re.search(r"(?:^|;\s*)overridable(?:;|$)", body))
overrides_field = body.split("rawOverrides={", 1)[1].split("}", 1)[0]
for entry in overrides_field.split(","):
package_name, separator, raw_value = entry.strip().partition("=")
if not separator:
continue
value: bool | int | str
if raw_value in ("true", "false"):
value = raw_value == "true"
else:
try:
value = int(raw_value)
except ValueError:
value = raw_value
self.results.append(
{
"change_id": int(match.group(1)),
"change_name": name_match.group(1) if name_match else None,
"change_state": state,
"overridable": overridable,
"package_name": package_name,
"override_value": value,
}
)
+39 -108
View File
@@ -4,6 +4,7 @@
# https://license.mvt.re/1.1/
from .artifact import AndroidArtifact
from .package_resolvers import parse_resolver_table
INTENT_NEW_OUTGOING_SMS = "android.provider.Telephony.NEW_OUTGOING_SMS"
INTENT_SMS_RECEIVED = "android.provider.Telephony.SMS_RECEIVED"
@@ -18,115 +19,45 @@ class DumpsysReceiversArtifact(AndroidArtifact):
"""
def check_indicators(self) -> None:
for intent, receivers in self.results.items():
for receiver in receivers:
if intent == INTENT_NEW_OUTGOING_SMS:
self.log.info(
'Found a receiver to intercept outgoing SMS messages: "%s"',
receiver["receiver"],
)
elif intent == INTENT_SMS_RECEIVED:
self.log.info(
'Found a receiver to intercept incoming SMS messages: "%s"',
receiver["receiver"],
)
elif intent == INTENT_DATA_SMS_RECEIVED:
self.log.info(
'Found a receiver to intercept incoming data SMS message: "%s"',
receiver["receiver"],
)
elif intent == INTENT_PHONE_STATE:
self.log.info(
"Found a receiver monitoring "
'telephony state/incoming calls: "%s"',
receiver["receiver"],
)
elif intent == INTENT_NEW_OUTGOING_CALL:
self.log.info(
'Found a receiver monitoring outgoing calls: "%s"',
receiver["receiver"],
)
for receiver in self.results:
intent = receiver["key"]
if intent == INTENT_NEW_OUTGOING_SMS:
self.log.info(
'Found a receiver to intercept outgoing SMS messages: "%s"',
receiver["component"],
)
elif intent == INTENT_SMS_RECEIVED:
self.log.info(
'Found a receiver to intercept incoming SMS messages: "%s"',
receiver["component"],
)
elif intent == INTENT_DATA_SMS_RECEIVED:
self.log.info(
'Found a receiver to intercept incoming data SMS message: "%s"',
receiver["component"],
)
elif intent == INTENT_PHONE_STATE:
self.log.info(
'Found a receiver monitoring telephony state/incoming calls: "%s"',
receiver["component"],
)
elif intent == INTENT_NEW_OUTGOING_CALL:
self.log.info(
'Found a receiver monitoring outgoing calls: "%s"',
receiver["component"],
)
if not self.indicators:
continue
if not self.indicators:
continue
ioc_match = self.indicators.check_app_id(receiver["package_name"])
if ioc_match:
self.alertstore.critical(
ioc_match.message,
"",
{intent: receiver},
matched_indicator=ioc_match.ioc,
)
continue
ioc_match = self.indicators.check_app_id(receiver["package_name"])
if ioc_match:
self.alertstore.critical(
ioc_match.message,
"",
receiver,
matched_indicator=ioc_match.ioc,
)
def parse(self, output: str) -> None:
self.results: dict[str, list[dict[str, str]]] = {}
in_receiver_resolver_table = False
in_non_data_actions = False
intent = None
for line in output.splitlines():
if line.startswith("Receiver Resolver Table:"):
in_receiver_resolver_table = True
continue
if not in_receiver_resolver_table:
continue
if line.startswith(" Non-Data Actions:"):
in_non_data_actions = True
continue
if not in_non_data_actions:
continue
# If we hit an empty line, the Non-Data Actions section should be
# finished.
if line.strip() == "":
break
# We detect the action name.
if (
line.startswith(" " * 6)
and not line.startswith(" " * 8)
and ":" in line
):
intent = line.strip().replace(":", "")
self.results[intent] = []
continue
parts = line.strip().split(" ")
if len(parts) < 2:
# A single-token line here is not a receiver. Real dumpstate
# output can print an action header mis-indented (observed with
# 15 leading spaces instead of 6), which used to raise
# IndexError and abort the whole module. Treat a trailing-colon
# token as the next action, skip anything else.
if parts[0].endswith(":"):
intent = parts[0][:-1]
self.results.setdefault(intent, [])
continue
# If we are not in an intent block yet, skip.
if not intent:
continue
# If we are in a block but the line does not start with 8 spaces
# it means the block ended a new one started, so we reset and
# continue.
if not line.startswith(" " * 8):
intent = None
continue
# If we got this far, we are processing receivers for the
# activities we are interested in.
receiver = parts[1]
package_name = receiver.split("/")[0]
self.results[intent].append(
{
"package_name": package_name,
"receiver": receiver,
}
)
self.results = parse_resolver_table(output, "Receiver")
+9 -12
View File
@@ -28,19 +28,16 @@ INTERESTING_PROPERTIES = [
class GetProp(AndroidArtifact):
def parse(self, entry: str) -> None:
self.results: List[Dict[str, str]] = []
rxp = re.compile(r"\[(.+?)\]: \[(.+?)\]")
# A property value may span several lines: persist.sys.boot.reason.history
# prints one boot per line. Matching the whole section instead of line by
# line lets a value run to the first closing bracket that ends a line.
rxp = re.compile(
r"^[ \t]*\[([^]]+)\]: \[(.*?)\][ \t\r]*$",
re.MULTILINE | re.DOTALL,
)
for line in entry.splitlines():
line = line.strip()
if line == "":
continue
matches = re.findall(rxp, line)
if not matches or len(matches[0]) != 2:
continue
prop_entry = {"name": matches[0][0], "value": matches[0][1]}
self.results.append(prop_entry)
for name, value in rxp.findall(entry):
self.results.append({"name": name, "value": value})
def get_device_timezone(self) -> str | None:
"""
+45
View File
@@ -117,6 +117,51 @@ class Mounts(AndroidArtifact):
# Skip lines that don't match expected format
continue
@staticmethod
def parse_mountinfo(entry: str, process_id: int) -> list[dict[str, Any]]:
"""Parse Linux /proc/PID/mountinfo records."""
results = []
for line in entry.splitlines():
fields = line.split()
if "-" not in fields:
continue
separator = fields.index("-")
if separator < 6 or len(fields) < separator + 4:
continue
try:
mount_id = int(fields[0])
parent_id = int(fields[1])
except ValueError:
continue
mount_options = fields[5].split(",")
super_options = fields[separator + 3].split(",")
options = list(dict.fromkeys(mount_options + super_options))
mount_point = fields[4].replace("\\040", " ")
device = fields[separator + 2].replace("\\040", " ")
filesystem_type = fields[separator + 1]
is_system = mount_point in SUSPICIOUS_MOUNT_POINTS or any(
mount_point.startswith(f"{prefix}/")
for prefix in SUSPICIOUS_MOUNT_POINTS
)
results.append(
{
"mount_id": mount_id,
"parent_id": parent_id,
"major_minor": fields[2],
"root": fields[3].replace("\\040", " "),
"mount_point": mount_point,
"device": device,
"filesystem_type": filesystem_type,
"mount_options": ",".join(options),
"options_list": options,
"optional_fields": fields[6:separator],
"is_system_partition": is_system,
"is_read_write": "rw" in options,
"process_ids": [process_id],
}
)
return results
def check_indicators(self) -> None:
"""
Check for suspicious mount configurations that may indicate root access
@@ -0,0 +1,70 @@
# 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 re
RESOLVER_TYPES = {
"Full MIME Types": "full_mime_type",
"Base MIME Types": "base_mime_type",
"Wild MIME Types": "wild_mime_type",
"Schemes": "scheme",
"Non-Data Actions": "non_data_action",
"MIME Typed Actions": "mime_typed_action",
}
ENTRY_RE = re.compile(
r"^\s{8,}[0-9a-fA-F]+\s+(?P<component>\S+)"
r"(?:\s+\((?P<filters>\d+)\s+filters?\))?\s*$"
)
def parse_resolver_table(content: str, table_name: str) -> list[dict]:
"""Parse every resolver category from a dumpsys package table."""
results: list[dict] = []
in_table = False
resolver_type: str | None = None
key: str | None = None
for line in content.splitlines():
if line.startswith(f"{table_name} Resolver Table:"):
in_table = True
continue
if not in_table:
continue
if line and not line.startswith(" "):
break
section_match = re.match(r"^ {2}([^ ].*):\s*$", line)
if section_match:
resolver_type = RESOLVER_TYPES.get(section_match.group(1))
key = None
continue
if resolver_type is None:
continue
key_match = re.match(r"^ {6,}([^ ].*):\s*$", line)
if key_match:
key = key_match.group(1)
continue
if key is None:
continue
entry_match = ENTRY_RE.match(line)
if not entry_match:
continue
component = entry_match.group("component")
results.append(
{
"resolver_type": resolver_type,
"key": key,
"package_name": component.split("/", 1)[0],
"component": component,
"filter_count": int(entry_match.group("filters") or 1),
}
)
return results
+73 -54
View File
@@ -1,72 +1,91 @@
# Mobile Verification Toolkit (MVT)
# Copyright (c) 2021-2023 The MVT Authors.
# Copyright (c) 2021-2026 The MVT Authors.
# Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/
from .artifact import AndroidArtifact
FIELD_NAMES = {
"LABEL": "label",
"USER": "user",
"PID": "pid",
"TID": "tid",
"PPID": "ppid",
"VSZ": "virtual_memory_size",
"RSS": "resident_set_size",
"WCHAN": "wchan",
"ADDR": "address",
"S": "state",
"PRI": "priority",
"NI": "nice",
"RTPRIO": "realtime_priority",
"SCH": "scheduler",
"PCY": "policy",
"TIME": "cpu_time",
"CMD": "command",
"NAME": "command",
}
INTEGER_FIELDS = {
"pid",
"tid",
"ppid",
"virtual_memory_size",
"resident_set_size",
"priority",
"nice",
}
class Processes(AndroidArtifact):
def parse(self, entry: str) -> None:
for line in entry.splitlines()[1:]:
proc = line.split()
self.results = []
lines = [line for line in entry.splitlines() if line.strip()]
if not lines:
return
headers = lines[0].split()
if not all(header in FIELD_NAMES for header in headers):
return
# Skip empty lines
if len(proc) == 0:
for line in lines[1:]:
values = line.split(None, len(headers) - 1)
if len(values) != len(headers):
continue
# Sometimes WCHAN is empty.
if len(proc) == 8:
proc = proc[:5] + [""] + proc[5:]
# Sometimes there is the security label.
if proc[0].startswith("u:r"):
label = proc[0]
proc = proc[1:]
else:
label = ""
# Sometimes there is no WCHAN.
if len(proc) < 9:
proc = proc[:5] + [""] + proc[5:]
self.results.append(
{
"user": proc[0],
"pid": int(proc[1]),
"ppid": int(proc[2]),
"virtual_memory_size": int(proc[3]),
"resident_set_size": int(proc[4]),
"wchan": proc[5],
"aprocress": proc[6],
"stat": proc[7],
"proc_name": proc[8].strip("[]"),
"label": label,
}
)
result = {}
valid = True
for header, raw in zip(headers, values):
key = FIELD_NAMES[header]
value: str | int = raw.strip("[]") if key == "command" else raw
if key in INTEGER_FIELDS:
try:
value = int(value)
except ValueError:
valid = False
break
result[key] = value
if valid:
self.results.append(result)
def check_indicators(self) -> None:
if not self.indicators:
return
for result in self.results:
proc_name = result.get("proc_name", "")
if not proc_name:
command = result.get("command", "")
if not isinstance(command, str):
continue
# Skipping this process because of false positives.
if result["proc_name"] == "gatekeeperd":
process_name = command.rsplit("/", 1)[-1]
if not process_name or process_name == "gatekeeperd":
continue
ioc_match = self.indicators.check_app_id(proc_name)
if ioc_match:
self.alertstore.critical(
ioc_match.message, "", result, matched_indicator=ioc_match.ioc
)
continue
ioc_match = self.indicators.check_process(proc_name)
if ioc_match:
self.alertstore.critical(
ioc_match.message, "", result, matched_indicator=ioc_match.ioc
)
for checker in (
self.indicators.check_app_id,
self.indicators.check_process,
):
ioc_match = checker(process_name)
if ioc_match:
self.alertstore.critical(
ioc_match.message,
"",
result,
matched_indicator=ioc_match.ioc,
)
break
+24
View File
@@ -3,6 +3,8 @@
# Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/
import re
from .artifact import AndroidArtifact
ANDROID_DANGEROUS_SETTINGS = [
@@ -60,6 +62,28 @@ ANDROID_DANGEROUS_SETTINGS = [
class Settings(AndroidArtifact):
def parse(self, content: str) -> None:
self.results: dict[str, dict[str, str]] = {}
namespace: str | None = None
for line in content.splitlines():
heading = re.match(
r"^(CONFIG|GLOBAL|SECURE|SYSTEM) SETTINGS \(user (\d+)\)$",
line.strip(),
)
if heading:
namespace = f"{heading.group(1).lower()}:user_{heading.group(2)}"
self.results[namespace] = {}
continue
if namespace is None or not line.startswith("_id:"):
continue
setting = re.match(
r"^_id:\S+\s+name:(.*?)\s+pkg:.*?\s+value:(.*?)"
r"(?:\s+default:.*\s+defaultSystemSet:(?:true|false))?$",
line,
)
if setting:
self.results[namespace][setting.group(1)] = setting.group(2)
def check_indicators(self) -> None:
for namespace, settings in self.results.items():
for key, value in settings.items():
+25 -7
View File
@@ -131,6 +131,13 @@ class TombstoneCrashArtifact(AndroidArtifact):
self, file_name: str, file_timestamp: datetime.datetime, data: bytes
) -> None:
"""Parse Android tombstone crash files from a protobuf object."""
self.results.append(self.parse_protobuf_record(file_name, file_timestamp, data))
def parse_protobuf_record(
self, file_name: str, file_timestamp: datetime.datetime, data: bytes
) -> dict:
if not data:
raise ValueError("empty protobuf tombstone")
tombstone_pb = Tombstone().parse(data)
tombstone_dict = tombstone_pb.to_dict(
casing=betterproto2.Casing.SNAKE, include_default_values=True
@@ -143,20 +150,31 @@ class TombstoneCrashArtifact(AndroidArtifact):
tombstone_dict["file_name"] = file_name
tombstone_dict["file_timestamp"] = convert_datetime_to_iso(file_timestamp)
tombstone_dict["process_name"] = self._proccess_name_from_thread(tombstone_dict)
if isinstance(tombstone_dict.get("selinux_label"), str):
tombstone_dict["selinux_label"] = tombstone_dict["selinux_label"].rstrip(
"\x00"
)
# Confirm the tombstone is valid, and matches the output model
tombstone = TombstoneCrashResult.model_validate(tombstone_dict)
self.results.append(tombstone.model_dump())
return tombstone.model_dump()
def parse(
self, file_name: str, file_timestamp: datetime.datetime, content: bytes
) -> None:
"""Parse text Android tombstone crash files."""
self.results.append(self.parse_text_record(file_name, file_timestamp, content))
def parse_text_record(
self, file_name: str, file_timestamp: datetime.datetime, content: bytes
) -> dict:
if not content:
raise ValueError("empty plaintext tombstone")
tombstone_dict = {
"file_name": file_name,
"file_timestamp": convert_datetime_to_iso(file_timestamp),
}
lines = content.decode("utf-8").splitlines()
lines = content.decode("utf-8", errors="replace").splitlines()
for line_num, line in enumerate(lines, 1):
if not line.strip() or TOMBSTONE_DELIMITER in line:
continue
@@ -171,7 +189,7 @@ class TombstoneCrashArtifact(AndroidArtifact):
# Validate the tombstone and add it to the results
tombstone = TombstoneCrashResult.model_validate(tombstone_dict)
self.results.append(tombstone.model_dump())
return tombstone.model_dump()
def _parse_tombstone_line(
self, line: str, key: str, destination_key: str, tombstone: dict
@@ -200,7 +218,9 @@ class TombstoneCrashArtifact(AndroidArtifact):
if not separator or line_key != key:
return False
value_clean = value.strip().strip("'")
value_clean = value.strip()
if len(value_clean) >= 2 and value_clean[0] == value_clean[-1] == "'":
value_clean = value_clean[1:-1]
if destination_key == "uid":
tombstone[destination_key] = int(value_clean)
elif destination_key == "process_uptime":
@@ -274,9 +294,7 @@ class TombstoneCrashArtifact(AndroidArtifact):
@staticmethod
def _parse_timestamp_string(timestamp: str) -> str:
timestamp_parsed = parser.parse(timestamp)
# Preserve the source wall-clock time while returning the project-wide ISO format.
local_timestamp = timestamp_parsed.replace(tzinfo=datetime.timezone.utc)
return convert_datetime_to_iso(local_timestamp)
return convert_datetime_to_iso(timestamp_parsed)
@staticmethod
def _proccess_name_from_thread(tombstone_dict: dict) -> str:
@@ -3,6 +3,8 @@
# 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 MVTModule
from .aqf_files import AQFFiles
from .aqf_getprop import AQFGetProp
from .aqf_log_timestamps import AQFLogTimestamps
@@ -12,7 +14,7 @@ from .aqf_settings import AQFSettings
from .mounts import Mounts
from .root_binaries import RootBinaries
ANDROIDQF_MODULES = [
ANDROIDQF_MODULES: list[type[MVTModule]] = [
AQFPackages,
AQFProcesses,
AQFGetProp,
+3 -1
View File
@@ -3,6 +3,8 @@
# 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 MVTModule
from .sms import SMS
BACKUP_MODULES = [SMS]
BACKUP_MODULES: list[type[MVTModule]] = [SMS]
@@ -3,6 +3,8 @@
# 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 MVTModule
from .dumpsys_accessibility import DumpsysAccessibility
from .dumpsys_activities import DumpsysActivities
from .dumpsys_appops import DumpsysAppops
@@ -16,8 +18,11 @@ from .dumpsys_receivers import DumpsysReceivers
from .dumpsys_adb_state import DumpsysADBState
from .fs_timestamps import BugReportTimestamps
from .tombstones import Tombstones
from .mounts import Mounts
from .processes import Processes
from .settings import Settings
BUGREPORT_MODULES = [
BUGREPORT_MODULES: list[type[MVTModule]] = [
DumpsysAccessibility,
DumpsysActivities,
DumpsysAppops,
@@ -31,4 +36,7 @@ BUGREPORT_MODULES = [
DumpsysADBState,
BugReportTimestamps,
Tombstones,
Processes,
Settings,
Mounts,
]
+26 -1
View File
@@ -72,7 +72,11 @@ class BugReportModule(MVTModule):
if not self.extract_path:
raise ValueError("extract_path is not set")
joined = os.path.join(self.extract_path, file_path)
if not Path(joined).resolve().is_relative_to(Path(self.extract_path).resolve()):
if (
not Path(joined)
.resolve()
.is_relative_to(Path(self.extract_path).resolve())
):
raise ValueError("unsafe file_path")
handle = open(joined, "rb")
@@ -100,6 +104,27 @@ class BugReportModule(MVTModule):
return None
@staticmethod
def extract_command_section(content: str, heading: str) -> str:
"""Return a bugreport command section without consuming the next one.
Bugreport separators include timing text, so looking for a line equal to
``------`` is not sufficient and can accidentally feed the remainder of
dumpstate to a parser.
"""
lines: list[str] = []
in_section = False
for line in content.splitlines():
stripped = line.strip()
if not in_section:
if stripped.startswith(heading):
in_section = True
continue
if stripped.startswith("------"):
break
lines.append(line)
return "\n".join(lines)
def _get_file_modification_time(self, file_path: str) -> datetime.datetime:
if self.zip_archive:
file_timetuple = self.zip_archive.getinfo(file_path).date_time
@@ -49,9 +49,7 @@ class DumpsysAccessibility(DumpsysAccessibilityArtifact, BugReportModule):
self.parse(content)
for result in self.results:
self.log.info(
'Found installed accessibility service "%s"', result.get("service")
)
self.log.info('Found accessibility service "%s"', result.get("component"))
self.log.info(
"Identified a total of %d accessibility services", len(self.results)
@@ -44,21 +44,8 @@ class DumpsysGetProp(GetPropArtifact, BugReportModule):
)
return
lines = []
in_getprop = False
for line in content.decode(errors="ignore").splitlines():
if line.strip().startswith("------ SYSTEM PROPERTIES"):
in_getprop = True
continue
if not in_getprop:
continue
if line.strip() == "------":
break
lines.append(line)
self.parse("\n".join(lines))
section = self.extract_command_section(
content.decode(errors="ignore"), "------ SYSTEM PROPERTIES"
)
self.parse(section)
self.log.info("Extracted %d Android system properties", len(self.results))
@@ -50,7 +50,10 @@ class DumpsysPackages(DumpsysPackagesArtifact, BugReportModule):
for result in self.results:
dangerous_permissions_count = 0
for perm in result["permissions"]:
permissions = list(result["permissions"])
for user in result.get("users", []):
permissions.extend(user.get("permissions", []))
for perm in permissions:
if perm["name"] in DANGEROUS_PERMISSIONS:
dangerous_permissions_count += 1
@@ -13,7 +13,7 @@ from mvt.common.module_types import ModuleResults
class DumpsysPlatformCompat(DumpsysPlatformCompatArtifact, BugReportModule):
"""This module extracts details on uninstalled apps."""
"""This module extracts raw per-package compatibility overrides."""
def __init__(
self,
@@ -48,4 +48,4 @@ class DumpsysPlatformCompat(DumpsysPlatformCompatArtifact, BugReportModule):
)
self.parse(content)
self.log.info("Found %d uninstalled apps", len(self.results))
self.log.info("Found %d package compatibility overrides", len(self.results))
@@ -33,7 +33,7 @@ class DumpsysReceivers(DumpsysReceiversArtifact, BugReportModule):
results=results,
)
self.results = results if results else {}
self.results = results if results else []
def run(self) -> None:
content = self._get_dumpstate_file()
@@ -49,4 +49,4 @@ class DumpsysReceivers(DumpsysReceiversArtifact, BugReportModule):
)
self.parse(dumpsys_section)
self.log.info("Extracted receivers for %d intents", len(self.results))
self.log.info("Extracted %d package receivers", len(self.results))
@@ -4,12 +4,15 @@
# https://license.mvt.re/1.1/
import logging
import datetime
from typing import Optional
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from mvt.common.utils import convert_datetime_to_iso
from .base import BugReportModule
from mvt.common.module_types import ModuleResults
from mvt.android.artifacts.file_timestamps import FileTimestampsArtifact
from mvt.android.artifacts.getprop import GetProp
class BugReportTimestamps(FileTimestampsArtifact, BugReportModule):
@@ -38,15 +41,44 @@ class BugReportTimestamps(FileTimestampsArtifact, BugReportModule):
def run(self) -> None:
filesystem_files = self._get_files_by_pattern("FS/*")
timezone_name = None
dumpstate = self._get_dumpstate_file()
if dumpstate:
section = self.extract_command_section(
dumpstate.decode("utf-8", errors="replace"),
"------ SYSTEM PROPERTIES",
)
properties = GetProp()
properties.parse(section)
timezone_name = properties.get_device_timezone()
timezone = None
if timezone_name:
try:
timezone = ZoneInfo(timezone_name)
except ZoneInfoNotFoundError:
self.log.warning("Unknown device timezone %s", timezone_name)
self.results = []
for file in filesystem_files:
# Only the modification time is available in the zip file metadata.
# The timezone is the local timezone of the machine the phone.
modification_time = self._get_file_modification_time(file)
utc_time = None
if timezone is not None:
utc_time = convert_datetime_to_iso(
modification_time.replace(tzinfo=timezone).astimezone(
datetime.timezone.utc
)
)
self.results.append(
{
"path": file,
"modified_time": convert_datetime_to_iso(modification_time),
"modified_time_utc": utc_time,
"timezone": timezone_name,
"timestamp_source": (
"zip_metadata" if self.zip_archive else "filesystem_metadata"
),
}
)
@@ -0,0 +1,41 @@
# Mobile Verification Toolkit (MVT)
# Copyright (c) 2021-2026 The MVT Authors.
import re
from mvt.android.artifacts.mounts import Mounts as MountsArtifact
from .base import BugReportModule
class Mounts(MountsArtifact, BugReportModule):
"""Extract and deduplicate process mount namespaces from mountinfo."""
def run(self) -> None:
mount_files = self._get_files_by_pattern("FS/proc/*/mountinfo")
if not mount_files:
mount_files = self._get_files_by_pattern("*/proc/*/mountinfo")
unique: dict[tuple, dict] = {}
for file_path in mount_files:
pid_match = re.search(r"/proc/(\d+)/mountinfo$", file_path)
if not pid_match:
continue
entries = self.parse_mountinfo(
self._get_file_content(file_path).decode("utf-8", errors="replace"),
int(pid_match.group(1)),
)
for entry in entries:
identity = (
entry["major_minor"],
entry["root"],
entry["mount_point"],
entry["device"],
entry["filesystem_type"],
entry["mount_options"],
)
if identity in unique:
unique[identity]["process_ids"].extend(entry["process_ids"])
else:
unique[identity] = entry
self.results = list(unique.values())
self.log.info("Extracted %d unique mount records", len(self.results))
@@ -0,0 +1,22 @@
# Mobile Verification Toolkit (MVT)
# Copyright (c) 2021-2026 The MVT Authors.
from mvt.android.artifacts.processes import Processes as ProcessesArtifact
from .base import BugReportModule
class Processes(ProcessesArtifact, BugReportModule):
"""Extract the process and thread table from dumpstate."""
def run(self) -> None:
data = self._get_dumpstate_file()
if not data:
self.log.error("Unable to find dumpstate file")
return
section = self.extract_command_section(
data.decode("utf-8", errors="replace"),
"------ PROCESSES AND THREADS",
)
self.parse(section)
self.log.info("Identified %d running process threads", len(self.results))
@@ -0,0 +1,22 @@
# Mobile Verification Toolkit (MVT)
# Copyright (c) 2021-2026 The MVT Authors.
from mvt.android.artifacts.settings import Settings as SettingsArtifact
from .base import BugReportModule
class Settings(SettingsArtifact, BugReportModule):
"""Extract all SettingsProvider namespaces and users."""
def run(self) -> None:
data = self._get_dumpstate_file()
if not data:
self.log.error("Unable to find dumpstate file")
return
section = self.extract_dumpsys_section(
data.decode("utf-8", errors="replace"), "DUMP OF SERVICE settings:"
)
self.parse(section)
count = sum(len(settings) for settings in self.results.values())
self.log.info("Identified %d Android settings", count)
+68 -13
View File
@@ -43,21 +43,76 @@ class Tombstones(TombstoneCrashArtifact, BugReportModule):
)
return
for tombstone_file in sorted(tombstone_files):
tombstone_filename = tombstone_file.split("/")[-1]
modification_time = self._get_file_modification_time(tombstone_file)
tombstone_data = self._get_file_content(tombstone_file)
grouped: dict[str, dict[str, str]] = {}
for file_path in tombstone_files:
file_name = file_path.rsplit("/", 1)[-1]
source = "protobuf" if file_name.endswith(".pb") else "text"
crash_id = file_name.removesuffix(".pb")
grouped.setdefault(crash_id, {})[source] = file_path
try:
if tombstone_file.endswith(".pb"):
self.parse_protobuf(
tombstone_filename, modification_time, tombstone_data
for crash_id, paths in sorted(grouped.items()):
parsed_sources: dict[str, dict] = {}
source_records: dict[str, dict] = {}
for source in ("text", "protobuf"):
file_path = paths.get(source)
if file_path is None:
continue
file_name = file_path.rsplit("/", 1)[-1]
file_timestamp = self._get_file_modification_time(file_path)
source_info = {
"file_name": file_name,
"file_timestamp": file_timestamp.isoformat(),
"parsed": False,
"error": None,
"record": None,
}
try:
data = self._get_file_content(file_path)
if source == "protobuf":
record = self.parse_protobuf_record(
file_name, file_timestamp, data
)
else:
record = self.parse_text_record(file_name, file_timestamp, data)
source_info["parsed"] = True
source_info["record"] = record
source_records[source] = record
except Exception as exc:
source_info["error"] = str(exc)
self.log.error(
"Error parsing tombstone file %s: %s", file_path, exc
)
else:
self.parse(tombstone_filename, modification_time, tombstone_data)
except ValueError as e:
# Catch any exceptions raised during parsing or validation.
self.log.error(f"Error parsing tombstone file {tombstone_file}: {e}")
parsed_sources[source] = source_info
if not source_records:
continue
preferred = source_records.get("protobuf") or source_records["text"]
canonical = dict(preferred)
text_record = source_records.get("text")
if text_record:
for key, value in text_record.items():
if canonical.get(key) in (None, "", [], {}):
canonical[key] = value
differences = {}
protobuf_record = source_records.get("protobuf")
if text_record and protobuf_record:
for key in text_record.keys() & protobuf_record.keys():
if key in ("file_name", "file_timestamp"):
continue
if text_record[key] != protobuf_record[key]:
differences[key] = {
"text": text_record[key],
"protobuf": protobuf_record[key],
}
canonical.update(
{
"crash_id": crash_id,
"sources": parsed_sources,
"differences": differences,
}
)
self.results.append(canonical)
self.log.info(
"Extracted a total of %d tombstone files",
@@ -3,11 +3,13 @@
# 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 MVTModule
from .connect_event import ConnectEvent
from .dns_event import DnsEvent
from .security_event import SecurityEvent
INTRUSION_LOGS_MODULES = [
INTRUSION_LOGS_MODULES: list[type[MVTModule]] = [
DnsEvent,
ConnectEvent,
SecurityEvent,
+1
View File
@@ -39,6 +39,7 @@ HELP_MSG_DECRYPT_BACKUP = "Decrypt an encrypted iTunes backup"
HELP_MSG_BACKUP_DESTINATION = (
"Path to the folder where the decrypted backup should be stored"
)
HELP_MSG_DECRYPT_JOBS = "Number of files to decrypt concurrently"
HELP_MSG_IOS_BACKUP_PASSWORD = (
"Password to use to decrypt the backup (or, set the {MVT_IOS_BACKUP_PASSWORD} "
"environment variable)"
+3 -1
View File
@@ -3,4 +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/
MVT_VERSION = "2026.7.29"
from importlib.metadata import version
MVT_VERSION = version("mvt")
+14 -2
View File
@@ -24,6 +24,7 @@ from mvt.common.help import (
HELP_MSG_VERSION,
HELP_MSG_DECRYPT_BACKUP,
HELP_MSG_BACKUP_DESTINATION,
HELP_MSG_DECRYPT_JOBS,
HELP_MSG_IOS_BACKUP_PASSWORD,
HELP_MSG_BACKUP_KEYFILE,
HELP_MSG_HASHES,
@@ -45,6 +46,10 @@ from mvt.common.help import (
HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK,
)
from mvt.common.password import prompt_password
from .decrypt_config import (
DEFAULT_DECRYPT_WORKERS,
MAX_DECRYPT_WORKERS,
)
# The commands import what they run only when they are invoked. This module is
# imported at every start of mvt-ios, including by shell completion on every
@@ -128,6 +133,13 @@ def version():
"decrypt-backup", context_settings=CONTEXT_SETTINGS, help=HELP_MSG_DECRYPT_BACKUP
)
@click.option("--destination", "-d", required=True, help=HELP_MSG_BACKUP_DESTINATION)
@click.option(
"--jobs",
type=click.IntRange(1, MAX_DECRYPT_WORKERS),
default=DEFAULT_DECRYPT_WORKERS,
show_default=True,
help=HELP_MSG_DECRYPT_JOBS,
)
@click.option(
"--password",
"-p",
@@ -146,10 +158,10 @@ def version():
@click.option("--hashes", "-H", is_flag=True, help=HELP_MSG_HASHES)
@click.argument("BACKUP_PATH", type=click.Path(exists=True))
@click.pass_context
def decrypt_backup(ctx, destination, password, key_file, hashes, backup_path):
def decrypt_backup(ctx, destination, jobs, password, key_file, hashes, backup_path):
from .decrypt import DecryptBackup
backup = DecryptBackup(backup_path, destination)
backup = DecryptBackup(backup_path, destination, max_workers=jobs)
if key_file:
if MVT_IOS_BACKUP_PASSWORD in os.environ:
+294 -76
View File
@@ -6,18 +6,179 @@
import binascii
import glob
import logging
import multiprocessing
import os
import os.path
import plistlib
import shutil
import sqlite3
import tempfile
from concurrent.futures import (
ALL_COMPLETED,
FIRST_COMPLETED,
Future,
ThreadPoolExecutor,
wait,
)
from pathlib import Path
from typing import Optional
from iOSbackup import iOSbackup
from iphone_backup_decrypt import EncryptedBackup
from iphone_backup_decrypt import google_iphone_dataprotection
from iphone_backup_decrypt.utils import FilePlist
from .decrypt_config import DEFAULT_DECRYPT_WORKERS, MAX_DECRYPT_WORKERS
log = logging.getLogger(__name__)
# Import pbkdf2_hmac from the same source iphone_backup_decrypt uses internally,
# so our key derivation is consistent with theirs.
try:
from fastpbkdf2 import pbkdf2_hmac
except ImportError:
import Crypto.Hash.SHA1
import Crypto.Hash.SHA256
import Crypto.Protocol.KDF
_HASH_FNS = {"sha1": Crypto.Hash.SHA1, "sha256": Crypto.Hash.SHA256}
def pbkdf2_hmac(hash_name, password, salt, iterations, dklen=None):
return Crypto.Protocol.KDF.PBKDF2(
password, salt, dklen, iterations, hmac_hash_module=_HASH_FNS[hash_name]
)
class MVTEncryptedBackup(EncryptedBackup):
"""Extends EncryptedBackup with derived key export/import.
NOTE: This subclass relies on internal APIs of iphone_backup_decrypt
(specifically _read_and_unlock_keybag, _keybag, and the Keybag class
internals). Pinned to iphone_backup_decrypt==0.9.0.
"""
def __init__(self, *, backup_directory, passphrase=None, derived_key=None):
if passphrase:
super().__init__(backup_directory=backup_directory, passphrase=passphrase)
self._derived_key = None # Will be set after keybag unlock
elif derived_key:
self._init_without_passphrase(backup_directory, derived_key)
else:
raise ValueError("Either passphrase or derived_key must be provided")
def _init_without_passphrase(self, backup_directory, derived_key):
"""Replicate parent __init__ state without requiring a passphrase."""
self.decrypted = False
self._backup_directory = os.path.expandvars(backup_directory)
self._passphrase = None
self._manifest_plist_path = os.path.join(
self._backup_directory, "Manifest.plist"
)
self._manifest_plist = None
self._manifest_db_path = os.path.join(self._backup_directory, "Manifest.db")
self._keybag = None
self._unlocked = False
self._temporary_folder = tempfile.mkdtemp()
self._temp_decrypted_manifest_db_path = os.path.join(
self._temporary_folder, "Manifest.db"
)
self._temp_manifest_db_conn = None
self._derived_key = derived_key # 32 raw bytes
def _read_and_unlock_keybag(self):
"""Override to capture derived key on password unlock, or use
a pre-derived key to skip PBKDF2."""
if self._unlocked:
return self._unlocked
with open(self._manifest_plist_path, "rb") as infile:
self._manifest_plist = plistlib.load(infile)
self._keybag = google_iphone_dataprotection.Keybag(
self._manifest_plist["BackupKeyBag"]
)
if self._derived_key:
# Skip PBKDF2, unwrap class keys directly with pre-derived key
self._unlocked = _unlock_keybag_with_derived_key(
self._keybag, self._derived_key
)
else:
# Normal path: full PBKDF2 derivation, capturing the intermediate key
self._unlocked, self._derived_key = _unlock_keybag_and_capture_key(
self._keybag, self._passphrase
)
self._passphrase = None
if not self._unlocked:
raise ValueError("Failed to decrypt keys: incorrect passphrase?")
return True
def get_decryption_key(self):
"""Return derived key as hex string (64 chars / 32 bytes)."""
if self._derived_key is None:
raise ValueError("No derived key available")
return self._derived_key.hex()
def extract_file_by_id(self, *, file_id, file_bplist, output_filename):
"""Extract one manifest entry without loading the whole file into memory."""
self._read_and_unlock_keybag()
file_plist = FilePlist(file_bplist)
if file_plist.encryption_key is None:
source_filename = os.path.join(
self._backup_directory, file_id[:2], file_id
)
shutil.copy2(source_filename, output_filename)
return
inner_key = self._keybag.unwrapKeyForClass(
file_plist.protection_class, file_plist.encryption_key
)
self._decrypt_file_to_disk(
file_id=file_id,
key=inner_key,
file_plist=file_plist,
output_filepath=output_filename,
)
def _unlock_keybag_with_derived_key(keybag, passphrase_key):
"""Unlock keybag class keys using a pre-derived passphrase_key,
skipping the expensive PBKDF2 rounds."""
WRAP_PASSPHRASE = 2
for classkey in keybag.classKeys.values():
if b"WPKY" not in classkey:
continue
if classkey[b"WRAP"] & WRAP_PASSPHRASE:
k = google_iphone_dataprotection._AESUnwrap(
passphrase_key, classkey[b"WPKY"]
)
if not k:
return False
classkey[b"KEY"] = k
return True
def _unlock_keybag_and_capture_key(keybag, passphrase):
"""Run full PBKDF2 key derivation and AES unwrap, returning
(success, passphrase_key) so the derived key can be exported."""
passphrase_round1 = pbkdf2_hmac(
"sha256", passphrase, keybag.attrs[b"DPSL"], keybag.attrs[b"DPIC"], 32
)
passphrase_key = pbkdf2_hmac(
"sha1", passphrase_round1, keybag.attrs[b"SALT"], keybag.attrs[b"ITER"], 32
)
WRAP_PASSPHRASE = 2
for classkey in keybag.classKeys.values():
if b"WPKY" not in classkey:
continue
if classkey[b"WRAP"] & WRAP_PASSPHRASE:
k = google_iphone_dataprotection._AESUnwrap(
passphrase_key, classkey[b"WPKY"]
)
if not k:
return False, None
classkey[b"KEY"] = k
return True, passphrase_key
class DecryptBackup:
"""This class provides functions to decrypt an encrypted iTunes backup
@@ -26,19 +187,68 @@ class DecryptBackup:
"""
def __init__(self, backup_path: str, dest_path: Optional[str] = None) -> None:
def __init__(
self,
backup_path: str,
dest_path: Optional[str] = None,
max_workers: int = DEFAULT_DECRYPT_WORKERS,
) -> None:
"""Decrypts an encrypted iOS backup.
:param backup_path: Path to the encrypted backup folder
:param dest_path: Path to the folder where to store the decrypted backup
"""
self.backup_path = os.path.abspath(backup_path)
self.dest_path = dest_path
self._backup = None
self._decryption_key = None
if not 1 <= max_workers <= MAX_DECRYPT_WORKERS:
raise ValueError(f"max_workers must be between 1 and {MAX_DECRYPT_WORKERS}")
self.max_workers = max_workers
self._backup: Optional[MVTEncryptedBackup] = None
self._decryption_key: Optional[str] = None
def can_process(self) -> bool:
return self._backup is not None
def _process_file(
self,
*,
file_id: str,
file_bplist: bytes,
output_path: Path,
relative_path: str,
domain: str,
) -> None:
assert self._backup is not None
self._backup.extract_file_by_id(
file_id=file_id,
file_bplist=file_bplist,
output_filename=str(output_path),
)
log.info(
"Decrypted file %s [%s] to %s/%s",
relative_path,
domain,
output_path.parent,
file_id,
)
@staticmethod
def _wait_for_files(
pending: dict[Future[None], str], *, all_files: bool = False
) -> None:
if not pending:
return
done, _ = wait(
pending,
return_when=ALL_COMPLETED if all_files else FIRST_COMPLETED,
)
for future in done:
relative_path = pending.pop(future)
try:
future.result()
except Exception as exc:
log.error("Failed to decrypt file %s: %s", relative_path, exc)
@staticmethod
def is_encrypted(backup_path: str) -> bool:
"""Query Manifest.db file to see if it's encrypted or not.
@@ -58,21 +268,6 @@ class DecryptBackup:
finally:
conn.close()
def _process_file(
self, relative_path: str, domain: str, item, file_id: str, item_folder: str
) -> None:
assert self._backup is not None
self._backup.getFileDecryptedCopy(
manifestEntry=item, targetName=file_id, targetFolder=item_folder
)
log.info(
"Decrypted file %s [%s] to %s/%s",
relative_path,
domain,
item_folder,
file_id,
)
def process_backup(self) -> None:
assert self._backup is not None
assert self.dest_path is not None
@@ -81,53 +276,71 @@ class DecryptBackup:
os.makedirs(self.dest_path)
manifest_path = os.path.join(self.dest_path, "Manifest.db")
# We extract a decrypted Manifest.db.
self._backup.getManifestDB()
# We store it to the destination folder.
shutil.copy(self._backup.manifestDB, manifest_path)
# Extract a decrypted Manifest.db to the destination folder.
self._backup.save_manifest_file(output_filename=manifest_path)
pool = multiprocessing.Pool(multiprocessing.cpu_count())
for item in self._backup.getBackupFilesList():
try:
file_id = item["backupFile"]
relative_path = item["relativePath"]
domain = item["domain"]
# This may be a partial backup. Skip files from the manifest
# which do not exist locally.
source_file_path = os.path.join(self.backup_path, file_id[0:2], file_id)
if not Path(source_file_path).resolve().is_relative_to(Path(self.backup_path).resolve()):
log.warning("Skipping unsafe file_id: %r", file_id)
continue
if not os.path.exists(source_file_path):
log.debug(
"Skipping file %s. File not found in encrypted backup directory.",
source_file_path,
)
continue
item_folder = os.path.join(self.dest_path, file_id[0:2]) # type: ignore[arg-type]
if not Path(os.path.join(item_folder, file_id)).resolve().is_relative_to(Path(self.dest_path).resolve()):
log.warning("Skipping unsafe file_id: %r", file_id)
continue
if not os.path.exists(item_folder):
os.makedirs(item_folder)
# iOSBackup getFileDecryptedCopy() claims to read a "file"
# parameter but the code actually is reading the "manifest" key.
# Add manifest plist to both keys to handle this.
item["manifest"] = item["file"]
pool.apply_async(
self._process_file,
args=(relative_path, domain, item, file_id, item_folder),
# Iterate over all files in the backup and decrypt them,
# preserving the XX/file_id directory structure that downstream
# modules expect.
backup_root = Path(self.backup_path).resolve()
dest_root = Path(self.dest_path).resolve()
pending: dict[Future[None], str] = {}
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
with self._backup.manifest_db_cursor() as cur:
cur.execute(
"SELECT fileID, domain, relativePath, file FROM Files WHERE flags=1"
)
except Exception as exc:
log.error("Failed to decrypt file %s: %s", relative_path, exc)
for file_id, domain, relative_path, file_bplist in cur:
# This may be a partial backup. Skip files from the manifest
# which do not exist locally.
source_file_path = backup_root / file_id[:2] / file_id
if not source_file_path.resolve().is_relative_to(backup_root):
log.warning("Skipping unsafe file_id: %r", file_id)
continue
if not os.path.exists(source_file_path):
log.debug(
"Skipping file %s. File not found in encrypted "
"backup directory.",
source_file_path,
)
continue
pool.close()
pool.join()
output_path = dest_root / file_id[:2] / file_id
if not output_path.resolve().is_relative_to(dest_root):
log.warning("Skipping unsafe file_id: %r", file_id)
continue
output_path.parent.mkdir(parents=True, exist_ok=True)
if self.max_workers == 1:
try:
self._process_file(
file_id=file_id,
file_bplist=file_bplist,
output_path=output_path,
relative_path=relative_path,
domain=domain,
)
except Exception as exc:
log.error(
"Failed to decrypt file %s: %s",
relative_path,
exc,
)
continue
future = executor.submit(
self._process_file,
file_id=file_id,
file_bplist=file_bplist,
output_path=output_path,
relative_path=relative_path,
domain=domain,
)
pending[future] = relative_path
if len(pending) >= self.max_workers:
self._wait_for_files(pending)
self._wait_for_files(pending, all_files=True)
# Copying over the root plist files as well.
for file_name in os.listdir(self.backup_path):
@@ -168,20 +381,23 @@ class DecryptBackup:
return
try:
self._backup = iOSbackup(
udid=os.path.basename(self.backup_path),
cleartextpassword=password,
backuproot=os.path.dirname(self.backup_path),
self._backup = MVTEncryptedBackup(
backup_directory=self.backup_path,
passphrase=password,
)
# Eagerly trigger keybag unlock so wrong-password errors
# surface here rather than later during process_backup().
self._backup.test_decryption()
except Exception as exc:
self._backup = None
if (
isinstance(exc, KeyError)
and len(exc.args) > 0
and exc.args[0] == b"KEY"
isinstance(exc, ValueError)
and "passphrase" in str(exc).lower()
):
log.critical("Failed to decrypt backup. Password is probably wrong.")
elif (
isinstance(exc, FileNotFoundError)
and hasattr(exc, "filename")
and os.path.basename(exc.filename) == "Manifest.plist"
):
log.critical(
@@ -224,12 +440,14 @@ class DecryptBackup:
try:
key_bytes_raw = binascii.unhexlify(key_bytes)
self._backup = iOSbackup(
udid=os.path.basename(self.backup_path),
derivedkey=key_bytes_raw,
backuproot=os.path.dirname(self.backup_path),
self._backup = MVTEncryptedBackup(
backup_directory=self.backup_path,
derived_key=key_bytes_raw,
)
# Eagerly trigger keybag unlock so wrong-key errors surface here.
self._backup.test_decryption()
except Exception as exc:
self._backup = None
log.exception(exc)
log.critical(
"Failed to decrypt backup. Did you provide the correct key file?"
@@ -240,7 +458,7 @@ class DecryptBackup:
if not self._backup:
return
self._decryption_key = self._backup.getDecryptionKey()
self._decryption_key = self._backup.get_decryption_key()
log.info(
'Derived decryption key for backup at path %s is: "%s"',
self.backup_path,
+9
View File
@@ -0,0 +1,9 @@
# Mobile Verification Toolkit (MVT)
# Copyright (c) 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/
"""Configuration shared by the iOS CLI and backup decryption implementation."""
DEFAULT_DECRYPT_WORKERS = 4
MAX_DECRYPT_WORKERS = 32
+8 -1
View File
@@ -3,9 +3,16 @@
# 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 MVTModule
from .backup_info import BackupInfo
from .configuration_profiles import ConfigurationProfiles
from .manifest import Manifest
from .profile_events import ProfileEvents
BACKUP_MODULES = [BackupInfo, ConfigurationProfiles, Manifest, ProfileEvents]
BACKUP_MODULES: list[type[MVTModule]] = [
BackupInfo,
ConfigurationProfiles,
Manifest,
ProfileEvents,
]
+3 -1
View File
@@ -3,6 +3,8 @@
# 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 MVTModule
from .analytics import Analytics
from .analytics_ios_versions import AnalyticsIOSVersions
from .cache_files import CacheFiles
@@ -15,7 +17,7 @@ from .webkit_indexeddb import WebkitIndexedDB
from .webkit_localstorage import WebkitLocalStorage
from .webkit_safariviewservice import WebkitSafariViewService
FS_MODULES = [
FS_MODULES: list[type[MVTModule]] = [
CacheFiles,
Filesystem,
Netusage,
+3 -1
View File
@@ -3,6 +3,8 @@
# 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 MVTModule
from .applications import Applications
from .calendar import Calendar
from .calls import Calls
@@ -28,7 +30,7 @@ from .webkit_session_resource_log import WebkitSessionResourceLog
from .whatsapp import Whatsapp
from .whatsapp_contacts import WhatsappContacts
MIXED_MODULES = [
MIXED_MODULES: list[type[MVTModule]] = [
Calls,
ChromeFavicon,
ChromeHistory,
@@ -23,7 +23,7 @@ class TestDumpsysAccessibilityArtifact:
assert len(da.results) == 4
assert da.results[0]["package_name"] == "com.android.settings"
assert (
da.results[0]["service"]
da.results[0]["component"]
== "com.android.settings/com.samsung.android.settings.development.gpuwatch.GPUWatchInterceptor"
)
@@ -37,7 +37,9 @@ class TestDumpsysAccessibilityArtifact:
da.parse(data)
assert len(da.results) == 1
assert da.results[0]["package_name"] == "com.malware.accessibility"
assert da.results[0]["service"] == "com.malware.service.malwareservice"
assert da.results[0]["service_name"] == "com.malware.service.malwareservice"
assert da.results[0]["enabled"] is True
assert da.results[0]["installed"] is False
def test_accessibility_service_alert(self):
da = DumpsysAccessibilityArtifact()
@@ -52,6 +54,22 @@ class TestDumpsysAccessibilityArtifact:
assert da.alertstore.alerts[0].level == AlertLevel.MEDIUM
assert da.alertstore.alerts[0].event == da.results[0]
def test_same_component_is_kept_for_each_user(self):
da = DumpsysAccessibilityArtifact()
da.parse(
"""User state[attributes:{id=0
installed services: {
0 : com.example/.Service
}
User state[attributes:{id=10
installed services: {
0 : com.example/.Service
}
"""
)
assert [result["user_id"] for result in da.results] == [0, 10]
def test_ioc_check(self, indicator_file):
da = DumpsysAccessibilityArtifact()
file = get_artifact("android_data/dumpsys_accessibility.txt")
@@ -3,6 +3,8 @@
# 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
from mvt.android.artifacts.dumpsys_adb import DumpsysADBArtifact
from mvt.android.modules.bugreport.dumpsys_adb_state import DumpsysADBState
from mvt.common.alerts import AlertLevel
@@ -11,6 +13,17 @@ from ..utils import get_artifact
class TestDumpsysADBArtifact:
def test_parsing_binary_xml_recovers_key(self):
public_key = base64.b64encode(bytes(range(256)))
keystore = DumpsysADBArtifact().parse_binary_xml(
b"ABX\x00binary-key=" + public_key + b" user@host\x00lastConnection"
)
assert len(keystore) == 1
assert keystore[0]["key"] == public_key.decode()
assert keystore[0]["user"] == "user@host"
assert keystore[0]["last_connected"] is None
def test_parsing(self):
da_adb = DumpsysADBArtifact()
file = get_artifact("android_data/dumpsys_adb.txt")
+28 -1
View File
@@ -25,10 +25,37 @@ class TestDumpsysAppopsArtifact:
assert da.results[0]["uid"] == "0"
assert len(da.results[0]["permissions"]) == 1
assert da.results[0]["permissions"][0]["name"] == "MANAGE_IPSEC_TUNNELS"
assert da.results[0]["permissions"][0]["access"] == "allow"
assert da.results[0]["permissions"][0]["mode"] == "allow"
assert da.results[6]["package_name"] == "com.sec.factory.camera"
assert len(da.results[6]["permissions"][1]["entries"]) == 1
assert len(da.results[11]["permissions"]) == 4
wake_lock = next(
permission
for permission in da.results[11]["permissions"]
if permission["name"] == "WAKE_LOCK"
)
assert wake_lock["entries"][0]["duration"] == "+126ms"
def test_running_and_attribution_are_retained(self):
da = DumpsysAppopsArtifact()
da.parse(
""" Uid 0:
state=cch
Package com.example:
CAMERA (allow):
camera=[
Access: [fg-s] 2025-01-01 00:00:00.000 (-1s) duration=+2ms
]
RECORD_AUDIO (allow):
Running start at: +3s
"""
)
camera = da.results[0]["permissions"][0]["entries"][0]
running = da.results[0]["permissions"][1]["entries"][0]
assert camera["attribution"] == "camera"
assert running["event"] == "running"
assert running["relative_time"] == "+3s"
def test_ioc_check(self, indicator_file):
da = DumpsysAppopsArtifact()
@@ -57,18 +57,19 @@ class TestDumpsysBatteryDailyArtifact:
assert uninstall_alert.message == (
"Detected uninstall of package com.example.removed (vers 0)"
)
assert uninstall_alert.event_time == "2022-08-16"
assert uninstall_alert.event_time == "2022-08-16 15:56:39"
assert uninstall_alert.event["package_name"] == "com.example.removed"
assert uninstall_alert.event["vers"] == "0"
assert uninstall_alert.event["version_code"] == 0
assert uninstall_alert.event["action"] == "uninstall"
assert downgrade_alert.level == AlertLevel.MEDIUM
assert downgrade_alert.message == (
"Detected downgrade of package com.example.app from vers 10 to vers 9"
)
assert downgrade_alert.event_time == "2022-08-17"
assert downgrade_alert.event_time == "2022-08-17 15:56:39"
assert downgrade_alert.event["package_name"] == "com.example.app"
assert downgrade_alert.event["action"] == "downgrade"
assert downgrade_alert.event["previous_vers"] == "10"
assert downgrade_alert.event["previous_version_code"] == 10
def test_newest_first_update_is_not_reported_as_downgrade(self):
dba = DumpsysBatteryDailyArtifact()
@@ -107,7 +108,19 @@ class TestDumpsysBatteryDailyArtifact:
assert downgrade_alert.event_time == "2026-01-10"
assert downgrade_alert.event["package_name"] == "com.example.app"
assert downgrade_alert.event["action"] == "downgrade"
assert downgrade_alert.event["previous_vers"] == "102"
assert downgrade_alert.event["previous_version_code"] == 102
def test_duplicate_updates_retain_occurrence_count(self):
dba = DumpsysBatteryDailyArtifact()
dba.parse(
""" Daily from 2026-01-10-01-02-03 to 2026-01-11-04-05-06:
Update com.example.app vers=12
Update com.example.app vers=12
"""
)
assert dba.results[0]["occurrences"] == 2
assert dba.results[0]["period_start"] == "2026-01-10 01:02:03"
def test_reinstall_after_uninstall_is_not_reported_as_downgrade(self):
dba = DumpsysBatteryDailyArtifact()
@@ -54,6 +54,7 @@ class TestDumpsysBatteryHistoryArtifact:
assert len(dba.results) == 2
assert dba.results[0] == {
"time_elapsed": "07-15 20:27:39.431",
"timestamp": "1900-07-15 20:27:39.431000",
"event": "start_job",
"uid": "u0a123",
"package_name": "com.example",
@@ -61,3 +62,23 @@ class TestDumpsysBatteryHistoryArtifact:
}
assert dba.results[1]["event"] == "end_job"
assert dba.results[1]["uid"] == "u0a123"
def test_wake_lock_without_component_is_retained(self):
dba = DumpsysBatteryHistoryArtifact()
dba.parse(
"Battery History:\n"
" 0 (2) 100 RESET:TIME: 2025-09-05-01-04-52-139\n"
' +1s (2) 100 +running +wake_lock=1000:"*alarm*:TIME_TICK"\n'
' +2s (2) 100 +running +wake_lock=u0a1:"*walarm*:com.whatsapp.MessageHandler.LOGOUT_ACTION"\n'
"\n"
)
assert [record["event"] for record in dba.results] == ["wake", "wake"]
assert dba.results[0]["package_name"] is None
assert dba.results[1]["package_name"] == "com.whatsapp"
def test_decorated_sync_job_uses_component_package(self):
dba = DumpsysBatteryHistoryArtifact()
dba.parse('+1s (2) 100 +job=u0a1:"@SyncManager@gmail-ls/com.google:android"\n')
assert dba.results[0]["package_name"] == "com.google"
+24 -1
View File
@@ -53,9 +53,32 @@ Connection pool for /data/user/0/com.example/databases/current.db:
assert dbi.results == [
{
"isodate": "07-15 20:27:39.431",
"timestamp": "07-15 20:27:39.431",
"pid": None,
"action": "executeForCursorWindow",
"duration_ms": 1,
"status": "succeeded",
"sql": "SELECT 1",
"path": "/data/user/0/com.example/databases/current.db",
"pool_path": "/data/user/0/com.example/databases/current.db",
"connection_number": None,
"is_primary": None,
}
]
def test_parses_operations_from_multiple_connections(self):
dbi = DumpsysDBInfoArtifact()
dbi.parse(
"""Connection pool for /data/example.db:
Connection #0:
isPrimaryConnection: true
Most recently executed operations:
0: [2025-01-01 00:00:00.000] execute took 1ms - succeeded, sql="SELECT 1", path=/data/example.db
Connection #1:
isPrimaryConnection: false
Most recently executed operations:
0: [2025-01-01 00:00:01.000] execute took 2ms - succeeded, sql="SELECT 2", path=/data/example.db
"""
)
assert [record["connection_number"] for record in dbi.results] == [0, 1]
@@ -21,12 +21,16 @@ class TestDumpsysPackageActivitiesArtifact:
assert len(dpa.results) == 0
dpa.parse(data)
assert len(dpa.results) == 4
assert dpa.results[0]["package_name"] == "com.samsung.android.app.social"
assert len(dpa.results) == 10
assert dpa.results[0]["package_name"] == "com.samsung.android.messaging"
assert (
dpa.results[0]["activity"]
== "com.samsung.android.app.social/.feed.FeedsActivity"
dpa.results[0]["component"]
== "com.samsung.android.messaging/.ui.RcsTransferContent"
)
assert {result["resolver_type"] for result in dpa.results} == {
"full_mime_type",
"non_data_action",
}
def test_ioc_check(self, indicator_file):
dpa = DumpsysPackageActivitiesArtifact()
@@ -41,4 +45,4 @@ class TestDumpsysPackageActivitiesArtifact:
dpa.indicators = ind
assert len(dpa.alertstore.alerts) == 0
dpa.check_indicators()
assert len(dpa.alertstore.alerts) == 1
assert len(dpa.alertstore.alerts) == 2
+25 -36
View File
@@ -25,7 +25,11 @@ class TestDumpsysPackagesArtifact:
== "com.samsung.android.provider.filterprovider"
)
assert dpa.results[0]["version_name"] == "5.0.07"
assert dpa.results[0]["first_install_time"] == "2008-12-31 16:00:00"
assert dpa.results[0]["version_code"] == 500700000
assert dpa.results[0]["min_sdk"] == 28
assert dpa.results[0]["target_sdk"] == 28
assert dpa.results[0]["package_type"] == "active"
assert dpa.results[0]["users"][0]["user_id"] == 0
assert dpa.results[0]["system"] is True
def test_parsing_system_flag(self):
@@ -60,42 +64,27 @@ class TestDumpsysPackagesArtifact:
dpa.check_indicators()
assert len(dpa.alertstore.alerts) == 1
def test_per_user_fields_use_primary_user(self):
details = DumpsysPackagesArtifact.parse_dumpsys_package_for_details(
""" User 0: installed=true
firstInstallTime=2024-01-10 09:19:39
runtime permissions:
android.permission.CAMERA: granted=true
User 95: installed=false
firstInstallTime=1970-01-01 01:00:00
runtime permissions:
android.permission.CAMERA: granted=false
android.permission.RECORD_AUDIO: granted=false
def test_hidden_packages_and_per_user_state(self):
dpa = DumpsysPackagesArtifact()
dpa.parse(
"""Packages:
Package [com.example.active]:
appId=10001
versionCode=12 minSdk=29 targetSdk=35
User 0: installed=true hidden=false
firstInstallTime=2025-01-01 01:02:03
User 10: installed=false hidden=true
firstInstallTime=2025-01-02 01:02:03
Hidden system packages:
Package [com.example.hidden]:
appId=10002
installerPackageName=null
"""
)
assert details["first_install_time"] == "2024-01-10 09:19:39"
runtime_permissions = [
permission
for permission in details["permissions"]
if permission["type"] == "runtime"
assert [record["package_type"] for record in dpa.results] == [
"active",
"hidden_system",
]
assert runtime_permissions == [
{
"name": "android.permission.CAMERA",
"granted": True,
"type": "runtime",
}
]
def test_per_user_fields_fall_back_when_user_zero_is_missing(self):
details = DumpsysPackagesArtifact.parse_dumpsys_package_for_details(
""" User 10: installed=true
firstInstallTime=2024-02-10 09:19:39
runtime permissions:
android.permission.CAMERA: granted=true
"""
)
assert details["first_install_time"] == "2024-02-10 09:19:39"
assert details["permissions"][-1]["name"] == "android.permission.CAMERA"
assert len(dpa.results[0]["users"]) == 2
assert dpa.results[1]["installer"] is None
@@ -22,6 +22,9 @@ class TestDumpsysPlatformCompatArtifact:
assert len(dbi.results) == 2
assert dbi.results[0]["package_name"] == "org.torproject.torbrowser"
assert dbi.results[1]["package_name"] == "org.article19.circulo.next"
assert dbi.results[0]["change_id"] == 168419799
assert dbi.results[0]["change_name"] == "DOWNSCALED"
assert dbi.results[0]["override_value"] is False
def test_ioc_check(self, indicator_file):
dbi = DumpsysPlatformCompatArtifact()
@@ -19,17 +19,14 @@ class TestDumpsysReceiversArtifact:
assert len(dr.results) == 0
dr.parse(data)
assert len(dr.results) == 4
assert (
list(dr.results.keys())[0]
== "com.android.storagemanager.automatic.SHOW_NOTIFICATION"
)
assert (
dr.results["com.android.storagemanager.automatic.SHOW_NOTIFICATION"][0][
"package_name"
]
== "com.android.storagemanager"
assert len(dr.results) == 9
assert dr.results[0]["resolver_type"] == "full_mime_type"
storage_manager = next(
result
for result in dr.results
if result["key"] == "com.android.storagemanager.automatic.SHOW_NOTIFICATION"
)
assert storage_manager["package_name"] == "com.android.storagemanager"
def test_parsing_misindented_action(self):
dr = DumpsysReceiversArtifact()
@@ -44,12 +41,8 @@ Receiver Resolver Table:
dr.parse(data)
assert (
dr.results["android.intent.action.MY_PACKAGE_REPLACED"][0][
"package_name"
]
== "com.psycatgames.nhiegame"
)
assert dr.results[1]["key"] == "android.intent.action.MY_PACKAGE_REPLACED"
assert dr.results[1]["package_name"] == "com.psycatgames.nhiegame"
def test_parsing_misindented_first_action(self):
dr = DumpsysReceiversArtifact()
@@ -62,12 +55,8 @@ Receiver Resolver Table:
dr.parse(data)
assert (
dr.results["android.intent.action.MY_PACKAGE_REPLACED"][0][
"package_name"
]
== "com.psycatgames.nhiegame"
)
assert dr.results[0]["key"] == "android.intent.action.MY_PACKAGE_REPLACED"
assert dr.results[0]["package_name"] == "com.psycatgames.nhiegame"
def test_ioc_check(self, indicator_file):
dr = DumpsysReceiversArtifact()
+26
View File
@@ -39,3 +39,29 @@ class TestGetPropArtifact:
assert len(gp.alertstore.alerts) == 0
gp.check_indicators()
assert len(gp.alertstore.alerts) == 1
def test_empty_values_and_invalid_lines(self):
gp = GetProp()
gp.parse("[empty]: []\n[valid]: [value]\n0\n[broken]: [value")
assert gp.results == [
{"name": "empty", "value": ""},
{"name": "valid", "value": "value"},
]
def test_multiline_value(self):
gp = GetProp()
gp.parse(
"[persist.sys.boot.reason.history]: ["
"reboot,ota,1697044974\n"
"reboot,watchdog,1696958574]\n"
"[ro.build.version.sdk]: [35]\n"
)
assert gp.results == [
{
"name": "persist.sys.boot.reason.history",
"value": "reboot,ota,1697044974\nreboot,watchdog,1696958574",
},
{"name": "ro.build.version.sdk", "value": "35"},
]
+12 -1
View File
@@ -20,7 +20,7 @@ class TestProcessesArtifact:
assert len(p.results) == 0
p.parse(data)
assert len(p.results) == 17
assert p.results[0]["proc_name"] == "init"
assert p.results[0]["command"] == "init"
def test_ioc_check(self, indicator_file):
p = Processes()
@@ -36,3 +36,14 @@ class TestProcessesArtifact:
assert len(p.alertstore.alerts) == 0
p.check_indicators()
assert len(p.alertstore.alerts) == 1
def test_bugreport_thread_columns(self):
p = Processes()
p.parse(
"LABEL USER PID TID PPID VSZ RSS WCHAN ADDR S PRI NI RTPRIO SCH PCY TIME CMD\n"
"u:r:init:s0 root 1 2 0 100 20 0 0 S 19 0 - 0 fg 00:00:01 init\n"
)
assert p.results[0]["label"] == "u:r:init:s0"
assert p.results[0]["tid"] == 2
assert p.results[0]["command"] == "init"
+2 -5
View File
@@ -128,8 +128,5 @@ class TestTombstoneCrashArtifact:
assert tombstone_result.get("pid") == 25541
assert tombstone_result.get("process_name") == "mtk.ape.decoder"
# With Android logs we want to keep timestamps as device local time for consistency.
# We often don't know the time offset for a log entry and so can't convert everything to UTC.
# MVT should output the local time only:
# So original 2023-04-12 12:32:40.518290770+0200 -> 2023-04-12 12:32:40.000000
assert tombstone_result.get("timestamp") == "2023-04-12 12:32:40.518290"
# Tombstones include an explicit offset, so normalize them to UTC.
assert tombstone_result.get("timestamp") == "2023-04-12 10:32:40.518290"
+13
View File
@@ -13,6 +13,19 @@ from ..utils import get_android_androidqf, list_files
class TestAndroidqfMountsArtifact:
def test_parse_proc_mountinfo(self):
from mvt.android.artifacts.mounts import Mounts as MountsArtifact
results = MountsArtifact.parse_mountinfo(
"41 40 254:13 / / ro,relatime shared:1 - erofs /dev/block/dm-13 ro,seclabel\n",
123,
)
assert results[0]["mount_id"] == 41
assert results[0]["mount_point"] == "/"
assert results[0]["filesystem_type"] == "erofs"
assert results[0]["process_ids"] == [123]
def test_parse_mounts_token_checks(self):
"""
Test the artifact-level `parse` method using tolerant token checks.
+15
View File
@@ -6,12 +6,27 @@
from pathlib import Path
from mvt.android.modules.androidqf.aqf_settings import AQFSettings
from mvt.android.artifacts.settings import Settings
from mvt.common.module import run_module
from ..utils import get_android_androidqf, list_files
class TestSettingsModule:
def test_bugreport_settings_format(self):
settings = Settings()
settings.parse(
"GLOBAL SETTINGS (user 0)\n"
"_id:1 name:adb_wifi_enabled pkg:android value:0 default:0 defaultSystemSet:true\n"
"SECURE SETTINGS (user 10)\n"
"_id:2 name:accessibility_enabled pkg:android value:1\n"
)
assert settings.results == {
"global:user_0": {"adb_wifi_enabled": "0"},
"secure:user_10": {"accessibility_enabled": "1"},
}
def test_parsing(self):
data_path = get_android_androidqf()
m = AQFSettings(target_path=data_path)
+14 -8
View File
@@ -54,10 +54,11 @@ class TestBugreportAnalysis:
== "com.samsung.android.provider.filterprovider"
)
assert m.results[1]["package_name"] == "com.instagram.android"
assert m.results[0]["installer"] == ""
assert m.results[0]["installer"] is None
assert m.results[1]["installer"] == "com.android.vending"
assert len(m.results[0]["permissions"]) == 4
assert len(m.results[1]["permissions"]) == 32
assert len(m.results[1]["permissions"]) == 20
assert len(m.results[1]["users"][0]["permissions"]) == 19
def test_getprop_module(self):
m = self.launch_bug_report_module(DumpsysGetProp)
@@ -66,29 +67,34 @@ class TestBugreportAnalysis:
def test_receivers_match_exact_package_name(self, indicators_factory):
intent = "android.intent.action.PHONE_STATE"
false_positive = {
"resolver_type": "non_data_action",
"key": intent,
"package_name": "com.android.phone",
"receiver": (
"component": (
"com.android.phone/"
"com.android.services.telephony.sip.SipIncomingCallReceiver"
),
"filter_count": 1,
}
malicious_receiver = {
"resolver_type": "non_data_action",
"key": intent,
"package_name": "com.android.services",
"receiver": "com.android.services/com.example.SomeReceiver",
"component": "com.android.services/com.example.SomeReceiver",
"filter_count": 1,
}
module = DumpsysReceivers(
results={intent: [false_positive, malicious_receiver]}
)
module = DumpsysReceivers(results=[false_positive, malicious_receiver])
module.indicators = indicators_factory(app_ids=["com.android.services"])
module.check_indicators()
assert len(module.alertstore.alerts) == 1
alert = module.alertstore.alerts[0]
assert alert.event == {intent: malicious_receiver}
assert alert.event == malicious_receiver
assert alert.matched_indicator.value == "com.android.services"
def test_tombstones_modules(self):
m = self.launch_bug_report_module(Tombstones)
assert len(m.results) == 2
assert m.results[1]["pid"] == 3559
assert m.results[0]["sources"]["text"]["parsed"] is True
+191
View File
@@ -0,0 +1,191 @@
# 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 threading
from pathlib import Path
from Crypto.Cipher import AES
from mvt.ios.decrypt import DecryptBackup, MVTEncryptedBackup
def _encrypted_file(backup_path, file_id, key, plaintext):
padding_length = AES.block_size - (len(plaintext) % AES.block_size)
padded = plaintext + bytes([padding_length]) * padding_length
encrypted = AES.new(key, AES.MODE_CBC, iv=b"\x00" * AES.block_size).encrypt(
padded
)
source_path = backup_path / file_id[:2] / file_id
source_path.parent.mkdir(parents=True)
source_path.write_bytes(encrypted)
def test_extract_file_by_id_preserves_bytes_with_wrong_manifest_size(
mocker, tmp_path
):
file_id = "ab" + "1" * 38
plaintext = b"complete decrypted content"
inner_key = b"k" * 32
_encrypted_file(tmp_path, file_id, inner_key, plaintext)
file_plist = mocker.Mock(
encryption_key=b"wrapped-key",
protection_class=1,
filesize=1,
mtime=None,
)
mocker.patch("mvt.ios.decrypt.FilePlist", return_value=file_plist)
backup = MVTEncryptedBackup(
backup_directory=str(tmp_path), derived_key=b"d" * 32
)
mocker.patch.object(backup, "_read_and_unlock_keybag", return_value=True)
backup._keybag = mocker.Mock()
backup._keybag.unwrapKeyForClass.return_value = inner_key
streaming_decrypt = mocker.spy(backup, "_decrypt_file_to_disk")
output_path = tmp_path / "output"
backup.extract_file_by_id(
file_id=file_id,
file_bplist=b"plist",
output_filename=str(output_path),
)
assert output_path.read_bytes() == plaintext
streaming_decrypt.assert_called_once()
def test_extract_file_by_id_copies_unencrypted_files(mocker, tmp_path):
file_id = "cd" + "2" * 38
source_path = tmp_path / file_id[:2] / file_id
source_path.parent.mkdir(parents=True)
source_path.write_bytes(b"plain content")
file_plist = mocker.Mock(encryption_key=None)
mocker.patch("mvt.ios.decrypt.FilePlist", return_value=file_plist)
backup = MVTEncryptedBackup(
backup_directory=str(tmp_path), derived_key=b"d" * 32
)
mocker.patch.object(backup, "_read_and_unlock_keybag", return_value=True)
output_path = tmp_path / "output"
backup.extract_file_by_id(
file_id=file_id,
file_bplist=b"plist",
output_filename=str(output_path),
)
assert output_path.read_bytes() == b"plain content"
def test_process_backup_rejects_unsafe_file_ids_and_destinations(mocker, tmp_path):
backup_path = tmp_path / "backup"
destination = tmp_path / "destination"
outside = tmp_path / "outside"
backup_path.mkdir()
destination.mkdir()
outside.mkdir()
safe_file_id = "ef" + "3" * 38
unsafe_file_id = "../../outside-file"
symlink_file_id = "ab" + "4" * 38
for file_id in (safe_file_id, symlink_file_id):
source_path = backup_path / file_id[:2] / file_id
source_path.parent.mkdir(parents=True, exist_ok=True)
source_path.write_bytes(b"encrypted")
(destination / "ab").symlink_to(outside, target_is_directory=True)
cursor = mocker.MagicMock()
cursor.__iter__.return_value = iter(
[
(safe_file_id, "Domain", "safe", b"plist"),
(unsafe_file_id, "Domain", "unsafe", b"plist"),
(symlink_file_id, "Domain", "symlink", b"plist"),
]
)
cursor_context = mocker.MagicMock()
cursor_context.__enter__.return_value = cursor
backup = mocker.MagicMock()
backup.manifest_db_cursor.return_value = cursor_context
def extract_file_by_id(*, output_filename, **kwargs):
Path(output_filename).write_bytes(b"decrypted")
backup.extract_file_by_id.side_effect = extract_file_by_id
decryptor = DecryptBackup(
str(backup_path), str(destination), max_workers=1
)
decryptor._backup = backup
decryptor.process_backup()
assert (destination / safe_file_id[:2] / safe_file_id).read_bytes() == b"decrypted"
assert not (outside / symlink_file_id).exists()
backup.extract_file_by_id.assert_called_once()
assert backup.extract_file_by_id.call_args.kwargs["file_id"] == safe_file_id
def test_process_backup_decrypts_files_concurrently(mocker, tmp_path):
backup_path = tmp_path / "backup"
destination = tmp_path / "destination"
backup_path.mkdir()
file_ids = ["ab" + "1" * 38, "cd" + "2" * 38]
for file_id in file_ids:
source_path = backup_path / file_id[:2] / file_id
source_path.parent.mkdir()
source_path.write_bytes(b"encrypted")
cursor = mocker.MagicMock()
cursor.__iter__.return_value = iter(
(file_id, "Domain", file_id, b"plist") for file_id in file_ids
)
cursor_context = mocker.MagicMock()
cursor_context.__enter__.return_value = cursor
barrier = threading.Barrier(2)
backup = mocker.MagicMock()
backup.manifest_db_cursor.return_value = cursor_context
def extract_file_by_id(*, file_id, output_filename, **kwargs):
barrier.wait(timeout=5)
Path(output_filename).write_bytes(file_id.encode())
backup.extract_file_by_id.side_effect = extract_file_by_id
decryptor = DecryptBackup(str(backup_path), str(destination), max_workers=2)
decryptor._backup = backup
decryptor.process_backup()
for file_id in file_ids:
assert (destination / file_id[:2] / file_id).read_bytes() == file_id.encode()
def test_process_backup_logs_worker_errors(mocker, tmp_path, caplog):
backup_path = tmp_path / "backup"
destination = tmp_path / "destination"
backup_path.mkdir()
file_id = "ef" + "3" * 38
source_path = backup_path / file_id[:2] / file_id
source_path.parent.mkdir()
source_path.write_bytes(b"encrypted")
cursor = mocker.MagicMock()
cursor.__iter__.return_value = iter([(file_id, "Domain", "failing-file", b"plist")])
cursor_context = mocker.MagicMock()
cursor_context.__enter__.return_value = cursor
backup = mocker.MagicMock()
backup.manifest_db_cursor.return_value = cursor_context
backup.extract_file_by_id.side_effect = ValueError("broken file")
decryptor = DecryptBackup(str(backup_path), str(destination))
decryptor._backup = backup
with caplog.at_level(logging.ERROR, logger="mvt.ios.decrypt"):
decryptor.process_backup()
assert "Failed to decrypt file failing-file: broken file" in caplog.text
Generated
+8 -7
View File
@@ -624,16 +624,15 @@ wheels = [
]
[[package]]
name = "iosbackup"
version = "0.9.925"
name = "iphone-backup-decrypt"
version = "0.9.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "nskeyedunarchiver" },
{ name = "pycryptodome" },
]
sdist = { url = "https://files.pythonhosted.org/packages/db/b8/4cd52322deceb942b9e18b127d45d112c2f7a3ec7821ab528659d4f04275/iOSbackup-0.9.925.tar.gz", hash = "sha256:33545a9249e5b3faaadf1ee782fe6bdfcdb70fae0defba1acee336a65f93d1ca", size = 25228, upload-time = "2022-10-16T01:02:46.474Z" }
sdist = { url = "https://files.pythonhosted.org/packages/6f/e7/bcdacdec21d628122ba240e7f742ab2175149e58672be63af55ff37a0f28/iphone_backup_decrypt-0.9.0.tar.gz", hash = "sha256:13b18fef3c8e3af627914f8c1a429bbc5555dfb0505239ba49efe99984cc0c96", size = 16125, upload-time = "2024-09-18T15:50:12.179Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/e0/4e/8da5fdc2df642080c79f128b9412a6f33768a610a58d59fbdb7b4d019364/iOSbackup-0.9.925-py3-none-any.whl", hash = "sha256:348edab2b82499d55c17c4bfe02fc6ae7915a2cc8da2728893ed1e9ae61bcef3", size = 18326, upload-time = "2022-10-16T01:02:44.164Z" },
{ url = "https://files.pythonhosted.org/packages/b8/94/64a31be93f72e0a254bde68e4cf7d24aef37a0a985754a197fa1b028a665/iphone_backup_decrypt-0.9.0-py3-none-any.whl", hash = "sha256:55b5adfafac757f58aa6444b83a4cc2c20cdd699c6ff1d2f4b549936a5dad92c", size = 15767, upload-time = "2024-09-18T15:50:10.537Z" },
]
[[package]]
@@ -1022,11 +1021,12 @@ dependencies = [
{ name = "betterproto2" },
{ name = "click" },
{ name = "cryptography" },
{ name = "iosbackup" },
{ name = "iphone-backup-decrypt" },
{ name = "libusb1" },
{ name = "nskeyedunarchiver" },
{ name = "packaging" },
{ name = "pyahocorasick" },
{ name = "pycryptodome" },
{ name = "pydantic" },
{ name = "pydantic-settings" },
{ name = "python-dateutil" },
@@ -1068,11 +1068,12 @@ requires-dist = [
{ name = "betterproto2", specifier = "==0.10.0" },
{ name = "click", specifier = "==8.4.2" },
{ name = "cryptography", specifier = "==50.0.0" },
{ name = "iosbackup", specifier = "==0.9.925" },
{ name = "iphone-backup-decrypt", specifier = "==0.9.0" },
{ name = "libusb1", specifier = "==3.4.0" },
{ name = "nskeyedunarchiver", specifier = "==1.5.2" },
{ name = "packaging", specifier = "==26.3" },
{ name = "pyahocorasick", specifier = "==2.3.1" },
{ name = "pycryptodome", specifier = ">=3.20.0" },
{ name = "pydantic", specifier = "==2.13.4" },
{ name = "pydantic-settings", specifier = "==2.15.0" },
{ name = "python-dateutil", specifier = "==2.9.0.post0" },