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: steps:
- name: Checkout repository - name: Checkout repository
uses: actions/checkout@v7 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. # 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 - name: Log in to the Container registry
uses: docker/login-action@dbcb813823bdd20940b903addbd779551569679f # v4.6.0 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: pull_request:
branches: [ main ] branches: [ main ]
permissions:
contents: read
pull-requests: write # coverage comment
jobs: jobs:
build: build:
name: Run Python Tests name: Run Python Tests
@@ -13,6 +17,10 @@ jobs:
fail-fast: false fail-fast: false
matrix: matrix:
python-version: ['3.10', '3.11', '3.12', '3.13', '3.14'] 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: steps:
- uses: actions/checkout@v7 - uses: actions/checkout@v7
@@ -26,16 +34,23 @@ jobs:
python-version: ${{ matrix.python-version }} python-version: ${{ matrix.python-version }}
- name: Install Python dependencies - name: Install Python dependencies
run: | run: |
uv sync --locked --group dev --python ${{ matrix.python-version }} uv sync --locked --group dev
- name: Test with pytest - name: Test with pytest
run: | run: |
set -o pipefail set -o pipefail
make test-ci | tee pytest-coverage.txt 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 - name: Pytest coverage comment
continue-on-error: true # Workflows running on a fork can't post comments # One comment per PR, not one per matrix entry. PRs from forks get a
uses: MishaKav/pytest-coverage-comment@main # read-only token and can't post; the job summary above still works.
if: github.event_name == 'pull_request' if: github.event_name == 'pull_request' && matrix.python-version == '3.13'
continue-on-error: true
uses: MishaKav/pytest-coverage-comment@v1.12.2
with: with:
pytest-coverage-path: ./pytest-coverage.txt pytest-coverage-path: ./pytest-coverage.txt
junitxml-path: ./pytest.xml 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 \ binutils \
default-jre-headless \ default-jre-headless \
file \ file \
git \
jq \ jq \
less \ less \
libcurl4 \ libcurl4 \
-5
View File
@@ -32,8 +32,3 @@ clean:
dist: dist:
$(UV) build $(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", "simplejson==4.1.1",
"packaging==26.3", "packaging==26.3",
"appdirs==1.4.4", "appdirs==1.4.4",
"iOSbackup==0.9.925", "iphone_backup_decrypt==0.9.0",
"pycryptodome>=3.20.0",
"adb-shell[usb]==0.4.4", "adb-shell[usb]==0.4.4",
"libusb1==3.4.0", "libusb1==3.4.0",
"cryptography==50.0.0", "cryptography==50.0.0",
@@ -72,7 +73,7 @@ docs = [
] ]
[build-system] [build-system]
requires = ["setuptools>=61.0"] requires = ["setuptools>=61.0", "setuptools-scm>=8"]
build-backend = "setuptools.build_meta" build-backend = "setuptools.build_meta"
[tool.coverage.run] [tool.coverage.run]
@@ -121,5 +122,6 @@ where = ["src"]
[tool.setuptools.package-data] [tool.setuptools.package-data]
mvt = ["ios/data/*.json"] mvt = ["ios/data/*.json"]
[tool.setuptools.dynamic] [tool.setuptools_scm]
version = { attr = "mvt.common.version.MVT_VERSION" } # 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/ # https://license.mvt.re/1.1/
import re import re
from typing import Any
from .artifact import AndroidArtifact from .artifact import AndroidArtifact
@@ -20,10 +21,10 @@ class DumpsysAccessibilityArtifact(AndroidArtifact):
continue continue
self.alertstore.medium( self.alertstore.medium(
f'Found accessibility service: "{result["service"]}"', f'Found accessibility service: "{result["component"]}"',
"", "",
result, result,
) )
def parse(self, content: str) -> None: def parse(self, content: str) -> None:
""" """
@@ -33,41 +34,69 @@ class DumpsysAccessibilityArtifact(AndroidArtifact):
:param content: content of the accessibility section (string) :param content: content of the accessibility section (string)
""" """
# "Old" syntax self.results: list[dict[str, Any]] = []
in_services = False services: dict[tuple[int | None, str], dict] = {}
user_id: int | None = None
state: str | None = None
for line in content.splitlines(): for line in content.splitlines():
if line.strip().startswith("installed services:"): user_match = re.search(r"attributes:\{id=(\d+)", line)
in_services = True if user_match:
continue user_id = int(user_match.group(1))
if not in_services: stripped = line.strip()
continue state_match = re.match(
r"(?i)(installed|enabled|binding|bound|crashed) services\s*:\s*\{(.*)",
if line.strip() == "}": stripped,
# At end of installed services
break
service = line.split(":")[1].strip()
self.results.append(
{
"package_name": service.split("/")[0],
"service": service,
}
) )
if state_match:
# "New" syntax - AOSP >= 14 (?) state = state_match.group(1).lower()
# Looks like: inline = state_match.group(2)
# Enabled services:{{com.azure.authenticator/com.microsoft.brooklyn.module.accessibility.BrooklynAccessibilityService}, {com.agilebits.onepassword/com.agilebits.onepassword.filling.accessibility.FillingAccessibilityService}} for component in re.findall(
r"\{?([\w.$-]+/[\w.$-]+)(?:\s+\(A11yTool\))?\}?", inline
for line in content.splitlines(): ):
if line.strip().startswith("Enabled services:"): service = services.setdefault(
matches = re.finditer(r"{([^{]+?)}", line) (user_id, component), self._new_service(component, user_id)
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}
) )
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 base64
import binascii import binascii
import hashlib import hashlib
import re
from .artifact import AndroidArtifact from .artifact import AndroidArtifact
@@ -98,6 +99,34 @@ class DumpsysADBArtifact(AndroidArtifact):
return keystore 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 @staticmethod
def calculate_key_info(user_key: bytes) -> dict: def calculate_key_info(user_key: bytes) -> dict:
if b" " in user_key: if b" " in user_key:
@@ -118,7 +147,7 @@ class DumpsysADBArtifact(AndroidArtifact):
return { return {
"user": user.decode("utf-8"), "user": user.decode("utf-8"),
"fingerprint": key_fingerprint_colon, "fingerprint": key_fingerprint_colon,
"key": key_base64, "key": key_base64.decode("ascii", errors="replace"),
} }
def check_indicators(self) -> None: 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 # Keystore is in XML format on some devices and we need to parse it
if keystore_data and keystore_data.startswith(b"<?xml"): if keystore_data and keystore_data.startswith(b"<?xml"):
parsed["debugging_manager"]["keystore"] = self.parse_xml(keystore_data) 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: else:
# Keystore is not XML format # Keystore is not XML format
parsed["debugging_manager"]["keystore"] = keystore_data parsed["debugging_manager"]["keystore"] = keystore_data
parsed = parsed["debugging_manager"] 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 # Calculate key fingerprints for better readability
key_info = [] key_info = []
for user_key in parsed.get("user_keys", []): for user_key in parsed.get("user_keys", []):
+122 -81
View File
@@ -4,6 +4,7 @@
# https://license.mvt.re/1.1/ # https://license.mvt.re/1.1/
from datetime import datetime from datetime import datetime
import re
from typing import Any from typing import Any
from mvt.common.module_types import ModuleAtomicResult, ModuleSerializedResult from mvt.common.module_types import ModuleAtomicResult, ModuleSerializedResult
@@ -27,14 +28,14 @@ class DumpsysAppopsArtifact(AndroidArtifact):
continue continue
for entry in perm["entries"]: for entry in perm["entries"]:
if "timestamp" in entry: if entry.get("timestamp"):
records.append( records.append(
{ {
"timestamp": entry["timestamp"], "timestamp": entry["timestamp"],
"module": self.__class__.__name__, "module": self.__class__.__name__,
"event": entry["access"], "event": entry["event"],
"data": f"{result['package_name']} access to " "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 continue
# We use a placeholder entry to create a basic alert even without permission entries. # 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"]: for perm in result["permissions"]:
if ( if (
@@ -60,12 +61,12 @@ class DumpsysAppopsArtifact(AndroidArtifact):
): ):
for entry in sorted( for entry in sorted(
perm["entries"] or [placeholder_entry], perm["entries"] or [placeholder_entry],
key=lambda x: x["timestamp"], key=lambda x: x.get("timestamp") or "",
): ):
cleaned_result = result.copy() cleaned_result = result.copy()
cleaned_result["permissions"] = [perm] cleaned_result["permissions"] = [perm]
self.alertstore.medium( 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"], entry["timestamp"],
cleaned_result, cleaned_result,
) )
@@ -73,111 +74,151 @@ class DumpsysAppopsArtifact(AndroidArtifact):
elif result["package_name"] in RISKY_PACKAGES: elif result["package_name"] in RISKY_PACKAGES:
for entry in sorted( for entry in sorted(
perm["entries"] or [placeholder_entry], perm["entries"] or [placeholder_entry],
key=lambda x: x["timestamp"], key=lambda x: x.get("timestamp") or "",
): ):
cleaned_result = result.copy() cleaned_result = result.copy()
cleaned_result["permissions"] = [perm] cleaned_result["permissions"] = [perm]
self.alertstore.medium( 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"], entry["timestamp"],
cleaned_result, cleaned_result,
) )
def parse(self, output: str) -> None: def parse(self, output: str) -> None:
# self.results: List[Dict[str, Any]] = [] self.results: list[dict[str, Any]] = []
perm: dict[str, Any] = {} permission: dict[str, Any] | None = None
package: dict[str, Any] = {} package: dict[str, Any] | None = None
entry: dict[str, Any] = {} uid: str | None = None
uid = None uid_details: dict[str, Any] = {}
attribution: str | None = None
in_packages = False 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(): for line in output.splitlines():
if line.startswith(" Uid 0:"): uid_match = re.match(r"^ Uid ([^:]+):$", line)
if uid_match:
in_packages = True 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: if not in_packages:
continue continue
if line.startswith(" Uid "): uid_property = re.match(
uid = line[6:-1] r"^ (state|capability|appWidgetVisible)=(.*)$", line
if entry: )
perm["entries"].append(entry) if uid_property:
entry = {} key = {
if package: "state": "uid_state",
if perm: "appWidgetVisible": "app_widget_visible",
package["permissions"].append(perm) }.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 = {} default_mode = re.match(r"^ ([A-Z0-9_]+): mode=([^\s]+)", line)
self.results.append(package) if default_mode and package is None:
package = {} uid_details["default_modes"][default_mode.group(1)] = (
default_mode.group(2)
)
continue continue
if line.startswith(" Package "): if line.startswith(" Package "):
if entry: finish_package()
perm["entries"].append(entry)
entry = {}
if package:
if perm:
package["permissions"].append(perm)
perm = {}
self.results.append(package)
package = { package = {
"package_name": line[12:-1], "package_name": line[12:-1],
"permissions": [], "permissions": [],
"uid": uid, "uid": uid,
**uid_details,
} }
continue continue
if package and line.startswith(" ") and line[6] != " ": operation_match = re.match(
if entry: r"^ ([A-Z0-9_]+)(?: \(([^)]+)\))?:\s*$", line
perm["entries"].append(entry) )
entry = {} if package is not None and operation_match:
if perm: finish_permission()
package["permissions"].append(perm) permission = {
perm = {} "name": operation_match.group(1),
"mode": operation_match.group(2),
perm["name"] = line.split()[0] "entries": [],
perm["entries"] = [] }
if len(line.split()) > 1: attribution = None
perm["access"] = line.split()[1][1:-2]
continue continue
if line.startswith(" "): attribution_match = re.match(r"^\s{8,}([^=]+)=\[$", line)
# Permission entry like: if attribution_match:
# Reject: [fg-s]2021-05-19 22:02:52.054 (-314d1h25m2s33ms) attribution = attribution_match.group(1).strip()
access_type = line.split(":")[0].strip() continue
if access_type not in ["Access", "Reject"]: if line.strip() == "]":
# Skipping invalid access type. Some entries are not in the format we expect attribution = None
continue continue
if entry: if permission is None:
perm["entries"].append(entry) continue
entry = {} event_match = re.match(
r"^\s*(Access|Reject):\s*\[([^]]+)\]\s*"
entry["access"] = access_type r"(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d+)\s*"
entry["type"] = line[line.find("[") + 1 : line.find("]")] r"(\([^)]*\))?(?:\s+duration=([^\s]+))?",
line,
try: )
entry["timestamp"] = convert_datetime_to_iso( running_match = re.match(
datetime.strptime( r"^\s*Running start at:\s*(\S+(?: \S+)?)",
line[line.find("]") + 1 : line.find("(")].strip(), line,
"%Y-%m-%d %H:%M:%S.%f", )
) 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: permission["entries"].append(
# Invalid date format {
pass "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() == "": finish_package()
break
if entry:
perm["entries"].append(entry)
if perm:
package["permissions"].append(perm)
if package:
self.results.append(package)
@@ -18,18 +18,18 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact):
def serialize(self, record: ModuleAtomicResult) -> ModuleSerializedResult: def serialize(self, record: ModuleAtomicResult) -> ModuleSerializedResult:
action = record.get("action", "update") action = record.get("action", "update")
package_name = record["package_name"] 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)" data = f"Recorded uninstall of package {package_name} (vers 0)"
elif action == "downgrade": 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}" data = f"Recorded downgrade of package {package_name} from vers {prev_vers} to vers {vers}"
else: else:
data = f"Recorded update of package {package_name} with vers {vers}" data = f"Recorded update of package {package_name} with vers {vers}"
return { return {
"timestamp": record["from"], "timestamp": record["period_start"],
"module": self.__class__.__name__, "module": self.__class__.__name__,
"event": "battery_daily", "event": "battery_daily",
"data": data, "data": data,
@@ -48,6 +48,7 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact):
continue continue
def parse(self, output: str) -> None: def parse(self, output: str) -> None:
self.results = []
daily = None daily = None
daily_updates: list[dict[str, Any]] = [] daily_updates: list[dict[str, Any]] = []
records: list[dict[str, Any]] = [] records: list[dict[str, Any]] = []
@@ -59,7 +60,10 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact):
timeframe = line[13:].strip() timeframe = line[13:].strip()
date_from, date_to = timeframe.strip(":").split(" to ", 1) 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 continue
if not daily: if not daily:
@@ -70,21 +74,30 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact):
line = line.strip().replace("Update ", "") line = line.strip().replace("Update ", "")
package_name, vers = line.split(" ", 1) 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 already_seen = False
for update in daily_updates: 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 already_seen = True
break break
if not already_seen: if not already_seen:
update_record: dict[str, Any] = { update_record: dict[str, Any] = {
"action": "update", "action": "update",
"from": daily["from"], "period_start": daily["period_start"],
"to": daily["to"], "period_end": daily["period_end"],
"package_name": package_name, "package_name": package_name,
"vers": vers_nr, "version_code": version_code,
"occurrences": 1,
} }
daily_updates.append(update_record) daily_updates.append(update_record)
@@ -95,26 +108,31 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact):
self._detect_uninstalls_and_downgrades(records) self._detect_uninstalls_and_downgrades(records)
self.results.extend(records) self.results.extend(records)
def _detect_uninstalls_and_downgrades( @staticmethod
self, records: list[dict[str, Any]] def _format_daily_timestamp(value: str) -> str:
) -> None: 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] = {} package_versions: dict[str, int] = {}
for record in sorted( for record in sorted(
records, records,
key=lambda record: ( key=lambda record: (
record["from"], record["period_start"],
record["to"], record["period_end"],
record["package_name"], record["package_name"],
), ),
): ):
package_name = 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( self.alertstore.medium(
f"Detected uninstall of package {package_name} (vers 0)", f"Detected uninstall of package {package_name} (vers 0)",
record["from"], record["period_start"],
record, record,
) )
package_versions.pop(package_name, None) package_versions.pop(package_name, None)
@@ -128,11 +146,11 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact):
previous_vers = package_versions.get(package_name) previous_vers = package_versions.get(package_name)
if previous_vers is not None and current_vers < previous_vers: if previous_vers is not None and current_vers < previous_vers:
record["action"] = "downgrade" record["action"] = "downgrade"
record["previous_vers"] = str(previous_vers) record["previous_version_code"] = previous_vers
self.alertstore.medium( self.alertstore.medium(
f"Detected downgrade of package {package_name} " f"Detected downgrade of package {package_name} "
f"from vers {previous_vers} to vers {current_vers}", f"from vers {previous_vers} to vers {current_vers}",
record["from"], record["period_start"],
record, record,
) )
@@ -1,96 +1,177 @@
# Mobile Verification Toolkit (MVT) # 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 # Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/ # https://license.mvt.re/1.1/
import datetime
import re
from mvt.common.utils import convert_datetime_to_iso
from .artifact import AndroidArtifact from .artifact import AndroidArtifact
class DumpsysBatteryHistoryArtifact(AndroidArtifact): class DumpsysBatteryHistoryArtifact(AndroidArtifact):
""" """Parser for package-related events in dumpsys batterystats history."""
Parser for dumpsys dattery history events.
"""
def check_indicators(self) -> None: def check_indicators(self) -> None:
if not self.indicators: if not self.indicators:
return return
for result in self.results: 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: if ioc_match:
self.alertstore.critical( self.alertstore.critical(
ioc_match.message, "", result, matched_indicator=ioc_match.ioc 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: 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(): 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 continue
if not in_history:
if line.strip() == "": continue
if has_history_heading and not stripped:
break break
reset_match = re.search(r"(?:RESET:)?TIME:\s*(\S+)", stripped)
time_parts = line.strip().split() if reset_match:
time_elapsed = time_parts[0] parsed_time = self._parse_wall_time(reset_match.group(1))
if ( if parsed_time is not None:
len(time_parts) > 1 elapsed_token = stripped.split()[0]
and len(time_parts[0]) == 5 anchor_elapsed = self._elapsed_seconds(elapsed_token) or 0.0
and time_parts[0][2] == "-" anchor_time = parsed_time
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:
continue continue
self.results.append( fields = stripped.split()
{ if not fields:
"time_elapsed": time_elapsed, continue
"event": event, if len(fields) > 1 and re.fullmatch(r"\d{2}-\d{2}", fields[0]):
"uid": uid, time_elapsed = " ".join(fields[:2])
"package_name": package_name, line_time = self._parse_wall_time(time_elapsed)
"service": service, 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: def parse(self, output: str) -> None:
rxp = re.compile( rxp = re.compile(
r".*\[((?:[0-9]{4}-)?[0-9]{2}-[0-9]{2} " r"^\s*\d+:\s*\[((?:\d{4}-)?\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{3})\]\s*"
r"[0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3})\]\s*" r"(?:\[Pid:\((\d+)\)\])?\s*([\w-]+) took (\d+)ms - ([^,]+),"
r"(?:\[Pid:\((\d+)\)\])?([\w-]+).*?sql=\"(.+?)\"" r"\s*sql=\"(.*)\"(?:, path=(.*))?$"
) # pylint: disable=line-too-long )
pool = None pool: str | None = None
connection_number: int | None = None
is_primary: bool | None = None
in_operations = False in_operations = False
for line in output.splitlines(): for line in output.splitlines():
if line.startswith("Connection pool for "): if line.startswith("Connection pool for "):
pool = line.replace("Connection pool for ", "").rstrip(":") 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: if not pool:
continue continue
@@ -52,7 +63,6 @@ class DumpsysDBInfoArtifact(AndroidArtifact):
if not line.startswith(" "): if not line.startswith(" "):
in_operations = False in_operations = False
pool = None
continue continue
match = rxp.match(line) match = rxp.match(line)
@@ -60,11 +70,15 @@ class DumpsysDBInfoArtifact(AndroidArtifact):
continue continue
result = { result = {
"isodate": match.group(1), "timestamp": match.group(1),
"pid": int(match.group(2)) if match.group(2) else None,
"action": match.group(3), "action": match.group(3),
"sql": match.group(4), "duration_ms": int(match.group(4)),
"path": pool, "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) self.results.append(result)
@@ -4,6 +4,7 @@
# https://license.mvt.re/1.1/ # https://license.mvt.re/1.1/
from .artifact import AndroidArtifact from .artifact import AndroidArtifact
from .package_resolvers import parse_resolver_table
class DumpsysPackageActivitiesArtifact(AndroidArtifact): class DumpsysPackageActivitiesArtifact(AndroidArtifact):
@@ -19,67 +20,11 @@ class DumpsysPackageActivitiesArtifact(AndroidArtifact):
) )
continue continue
def parse(self, content: str): def parse(self, content: str) -> None:
""" """
Parse the Dumpsys Package section for activities Parse the Dumpsys Package section for activities
Adds results to self.results Adds results to self.results
:param content: content of the package section (string) :param content: content of the package section (string)
""" """
self.results = [] self.results = parse_resolver_table(content, "Activity")
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,
}
)
+205 -183
View File
@@ -1,10 +1,10 @@
# Mobile Verification Toolkit (MVT) # 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 # Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/ # https://license.mvt.re/1.1/
import re import re
from typing import Any, Dict, List, Optional from typing import Any
from mvt.android.utils import ROOT_PACKAGES from mvt.android.utils import ROOT_PACKAGES
from mvt.common.module_types import ModuleAtomicResult, ModuleSerializedResult from mvt.common.module_types import ModuleAtomicResult, ModuleSerializedResult
@@ -12,222 +12,244 @@ from mvt.common.module_types import ModuleAtomicResult, ModuleSerializedResult
from .artifact import AndroidArtifact 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): class DumpsysPackagesArtifact(AndroidArtifact):
def check_indicators(self) -> None: def check_indicators(self) -> None:
alerted_root_packages = set() alerted_root_packages = set()
for result in self.results: for result in self.results:
if result["package_name"] in ROOT_PACKAGES: package_name = result["package_name"]
if result["package_name"] in alerted_root_packages: if (
continue package_name in ROOT_PACKAGES
alerted_root_packages.add(result["package_name"]) and package_name not in alerted_root_packages
):
alerted_root_packages.add(package_name)
self.alertstore.medium( 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, result,
) )
continue
if not self.indicators: if not self.indicators:
continue continue
ioc_match = self.indicators.check_app_id(package_name)
ioc_match = self.indicators.check_app_id(result.get("package_name", ""))
if ioc_match: if ioc_match:
self.alertstore.critical( self.alertstore.critical(
ioc_match.message, "", result, matched_indicator=ioc_match.ioc ioc_match.message, "", result, matched_indicator=ioc_match.ioc
) )
def serialize(self, record: ModuleAtomicResult) -> ModuleSerializedResult: def serialize(self, record: ModuleAtomicResult) -> ModuleSerializedResult:
records = []
timestamps = [ 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": timestamp,
"timestamp": record["first_install_time"], "module": self.__class__.__name__,
}, "event": event,
{"event": "package_last_update", "timestamp": record["last_update_time"]}, "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 @staticmethod
def parse_dumpsys_package_for_details(output: str) -> Dict[str, Any]: def _permission(line: str, permission_type: str) -> dict:
""" name, _, details = line.strip().partition(":")
Parse one entry of a dumpsys package information granted_match = re.search(r"granted=(true|false)", details)
""" flags_match = re.search(r"flags=\[\s*([^]]*)\]", details)
details: Dict[str, Any] = { return {
"uid": "", "name": name,
"version_name": "", "type": permission_type,
"version_code": "", "granted": granted_match.group(1) == "true" if granted_match else None,
"timestamp": "", "flags": [
"first_install_time": "", flag.strip()
"last_update_time": "", for flag in (flags_match.group(1).split("|") if flags_match else [])
"installer": "", if flag.strip()
"system": False, ],
"permissions": list(),
"requested_permissions": list(),
} }
in_install_permissions = False
in_runtime_permissions = False @classmethod
in_declared_permissions = False def parse_dumpsys_package_for_details(cls, output: str) -> dict[str, Any]:
in_requested_permissions = True details: dict[str, Any] = {
current_user: Optional[int] = None "app_id": None,
first_install_times: Dict[Optional[int], str] = {} "version_name": None,
runtime_permissions: Dict[Optional[int], List[Dict[str, Any]]] = {} "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(): 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: 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: header = stripped.lower()
if line.startswith(" " * 4) and not line.startswith(" " * 6): if header in {
in_install_permissions = False "declared permissions:",
else: "install permissions:",
lineinfo = line.strip().split(":") "requested permissions:",
permission = lineinfo[0] "runtime permissions:",
granted = None }:
if "granted=" in lineinfo[1]: permission_section = header.split()[0]
granted = "granted=true" in lineinfo[1] continue
details["permissions"].append( if current_user is not None:
{"name": permission, "granted": granted, "type": "install"} user_property = re.match(
) r"(installReason|uninstallReason|dataDir|firstInstallTime)=(.*)",
if in_runtime_permissions: stripped,
if not line.startswith(" " * 8): )
in_runtime_permissions = False if user_property:
else: key = {
lineinfo = line.strip().split(":") "installReason": "install_reason",
permission = lineinfo[0] "uninstallReason": "uninstall_reason",
granted = None "dataDir": "data_dir",
if "granted=" in lineinfo[1]: "firstInstallTime": "first_install_time",
granted = "granted=true" in lineinfo[1] }[user_property.group(1)]
current_user[key] = _value(user_property.group(2))
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:
continue continue
package_name = matches[0] if permission_section == "requested" and line.startswith(" "):
package["package_name"] = package_name 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 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 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: def parse(self, content: str) -> None:
details = self.parse_dumpsys_package_for_details("\n".join(lines)) self.results: list[dict[str, Any]] = []
package.update(details) category: str | None = None
results.append(package) 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(): for line in content.splitlines():
if line.startswith("Packages:"): if line == "Packages:":
in_package_list = True finish()
category = "active"
continue continue
if line == "Hidden system packages:":
if not in_package_list: finish()
category = "hidden_system"
continue continue
package_match = re.match(r"^ Package \[([^]]+)\]", line)
if line.strip() == "": if package_match and category:
break finish()
package = {
package.append(line) "package_name": package_match.group(1),
"package_type": category,
self.results = self.parse_dumpsys_packages("\n".join(package)) }
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 # Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/ # https://license.mvt.re/1.1/
import re
from .artifact import AndroidArtifact from .artifact import AndroidArtifact
@@ -25,19 +27,37 @@ class DumpsysPlatformCompatArtifact(AndroidArtifact):
def parse(self, data: str) -> None: def parse(self, data: str) -> None:
for line in data.splitlines(): 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 continue
body = match.group(2)
if line.strip() == "": name_match = re.search(r"(?:^|;\s*)name=([^;]+)", body)
break state = (
"enabled"
# Look for rawOverrides field if re.search(r"(?:^|;\s*)enabled(?:;|$)", body)
if "rawOverrides={" in line: else "disabled"
# Extract the content inside the braces for rawOverrides )
overrides_field = line.split("rawOverrides={", 1)[1].split("};", 1)[0] overridable = bool(re.search(r"(?:^|;\s*)overridable(?:;|$)", body))
overrides_field = body.split("rawOverrides={", 1)[1].split("}", 1)[0]
for entry in overrides_field.split(", "): for entry in overrides_field.split(","):
# Extract app name package_name, separator, raw_value = entry.strip().partition("=")
uninstall_app = entry.split("=")[0].strip() if not separator:
continue
self.results.append({"package_name": uninstall_app}) 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/ # https://license.mvt.re/1.1/
from .artifact import AndroidArtifact from .artifact import AndroidArtifact
from .package_resolvers import parse_resolver_table
INTENT_NEW_OUTGOING_SMS = "android.provider.Telephony.NEW_OUTGOING_SMS" INTENT_NEW_OUTGOING_SMS = "android.provider.Telephony.NEW_OUTGOING_SMS"
INTENT_SMS_RECEIVED = "android.provider.Telephony.SMS_RECEIVED" INTENT_SMS_RECEIVED = "android.provider.Telephony.SMS_RECEIVED"
@@ -18,115 +19,45 @@ class DumpsysReceiversArtifact(AndroidArtifact):
""" """
def check_indicators(self) -> None: def check_indicators(self) -> None:
for intent, receivers in self.results.items(): for receiver in self.results:
for receiver in receivers: intent = receiver["key"]
if intent == INTENT_NEW_OUTGOING_SMS: if intent == INTENT_NEW_OUTGOING_SMS:
self.log.info( self.log.info(
'Found a receiver to intercept outgoing SMS messages: "%s"', 'Found a receiver to intercept outgoing SMS messages: "%s"',
receiver["receiver"], receiver["component"],
) )
elif intent == INTENT_SMS_RECEIVED: elif intent == INTENT_SMS_RECEIVED:
self.log.info( self.log.info(
'Found a receiver to intercept incoming SMS messages: "%s"', 'Found a receiver to intercept incoming SMS messages: "%s"',
receiver["receiver"], receiver["component"],
) )
elif intent == INTENT_DATA_SMS_RECEIVED: elif intent == INTENT_DATA_SMS_RECEIVED:
self.log.info( self.log.info(
'Found a receiver to intercept incoming data SMS message: "%s"', 'Found a receiver to intercept incoming data SMS message: "%s"',
receiver["receiver"], receiver["component"],
) )
elif intent == INTENT_PHONE_STATE: elif intent == INTENT_PHONE_STATE:
self.log.info( self.log.info(
"Found a receiver monitoring " 'Found a receiver monitoring telephony state/incoming calls: "%s"',
'telephony state/incoming calls: "%s"', receiver["component"],
receiver["receiver"], )
) elif intent == INTENT_NEW_OUTGOING_CALL:
elif intent == INTENT_NEW_OUTGOING_CALL: self.log.info(
self.log.info( 'Found a receiver monitoring outgoing calls: "%s"',
'Found a receiver monitoring outgoing calls: "%s"', receiver["component"],
receiver["receiver"], )
)
if not self.indicators: if not self.indicators:
continue continue
ioc_match = self.indicators.check_app_id(receiver["package_name"]) ioc_match = self.indicators.check_app_id(receiver["package_name"])
if ioc_match: if ioc_match:
self.alertstore.critical( self.alertstore.critical(
ioc_match.message, ioc_match.message,
"", "",
{intent: receiver}, receiver,
matched_indicator=ioc_match.ioc, matched_indicator=ioc_match.ioc,
) )
continue
def parse(self, output: str) -> None: def parse(self, output: str) -> None:
self.results: dict[str, list[dict[str, str]]] = {} self.results = parse_resolver_table(output, "Receiver")
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,
}
)
+9 -12
View File
@@ -28,19 +28,16 @@ INTERESTING_PROPERTIES = [
class GetProp(AndroidArtifact): class GetProp(AndroidArtifact):
def parse(self, entry: str) -> None: def parse(self, entry: str) -> None:
self.results: List[Dict[str, str]] = [] 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(): for name, value in rxp.findall(entry):
line = line.strip() self.results.append({"name": name, "value": value})
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)
def get_device_timezone(self) -> str | None: 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 # Skip lines that don't match expected format
continue 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: def check_indicators(self) -> None:
""" """
Check for suspicious mount configurations that may indicate root access 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) # 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 # Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/ # https://license.mvt.re/1.1/
from .artifact import AndroidArtifact 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): class Processes(AndroidArtifact):
def parse(self, entry: str) -> None: def parse(self, entry: str) -> None:
for line in entry.splitlines()[1:]: self.results = []
proc = line.split() 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 for line in lines[1:]:
if len(proc) == 0: values = line.split(None, len(headers) - 1)
if len(values) != len(headers):
continue continue
result = {}
# Sometimes WCHAN is empty. valid = True
if len(proc) == 8: for header, raw in zip(headers, values):
proc = proc[:5] + [""] + proc[5:] key = FIELD_NAMES[header]
value: str | int = raw.strip("[]") if key == "command" else raw
# Sometimes there is the security label. if key in INTEGER_FIELDS:
if proc[0].startswith("u:r"): try:
label = proc[0] value = int(value)
proc = proc[1:] except ValueError:
else: valid = False
label = "" break
result[key] = value
# Sometimes there is no WCHAN. if valid:
if len(proc) < 9: self.results.append(result)
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,
}
)
def check_indicators(self) -> None: def check_indicators(self) -> None:
if not self.indicators: if not self.indicators:
return return
for result in self.results: for result in self.results:
proc_name = result.get("proc_name", "") command = result.get("command", "")
if not proc_name: if not isinstance(command, str):
continue continue
process_name = command.rsplit("/", 1)[-1]
# Skipping this process because of false positives. if not process_name or process_name == "gatekeeperd":
if result["proc_name"] == "gatekeeperd":
continue continue
for checker in (
ioc_match = self.indicators.check_app_id(proc_name) self.indicators.check_app_id,
if ioc_match: self.indicators.check_process,
self.alertstore.critical( ):
ioc_match.message, "", result, matched_indicator=ioc_match.ioc ioc_match = checker(process_name)
) if ioc_match:
continue self.alertstore.critical(
ioc_match.message,
ioc_match = self.indicators.check_process(proc_name) "",
if ioc_match: result,
self.alertstore.critical( matched_indicator=ioc_match.ioc,
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 # Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/ # https://license.mvt.re/1.1/
import re
from .artifact import AndroidArtifact from .artifact import AndroidArtifact
ANDROID_DANGEROUS_SETTINGS = [ ANDROID_DANGEROUS_SETTINGS = [
@@ -60,6 +62,28 @@ ANDROID_DANGEROUS_SETTINGS = [
class Settings(AndroidArtifact): 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: def check_indicators(self) -> None:
for namespace, settings in self.results.items(): for namespace, settings in self.results.items():
for key, value in settings.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 self, file_name: str, file_timestamp: datetime.datetime, data: bytes
) -> None: ) -> None:
"""Parse Android tombstone crash files from a protobuf object.""" """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_pb = Tombstone().parse(data)
tombstone_dict = tombstone_pb.to_dict( tombstone_dict = tombstone_pb.to_dict(
casing=betterproto2.Casing.SNAKE, include_default_values=True casing=betterproto2.Casing.SNAKE, include_default_values=True
@@ -143,20 +150,31 @@ class TombstoneCrashArtifact(AndroidArtifact):
tombstone_dict["file_name"] = file_name tombstone_dict["file_name"] = file_name
tombstone_dict["file_timestamp"] = convert_datetime_to_iso(file_timestamp) tombstone_dict["file_timestamp"] = convert_datetime_to_iso(file_timestamp)
tombstone_dict["process_name"] = self._proccess_name_from_thread(tombstone_dict) 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 # Confirm the tombstone is valid, and matches the output model
tombstone = TombstoneCrashResult.model_validate(tombstone_dict) tombstone = TombstoneCrashResult.model_validate(tombstone_dict)
self.results.append(tombstone.model_dump()) return tombstone.model_dump()
def parse( def parse(
self, file_name: str, file_timestamp: datetime.datetime, content: bytes self, file_name: str, file_timestamp: datetime.datetime, content: bytes
) -> None: ) -> None:
"""Parse text Android tombstone crash files.""" """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 = { tombstone_dict = {
"file_name": file_name, "file_name": file_name,
"file_timestamp": convert_datetime_to_iso(file_timestamp), "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): for line_num, line in enumerate(lines, 1):
if not line.strip() or TOMBSTONE_DELIMITER in line: if not line.strip() or TOMBSTONE_DELIMITER in line:
continue continue
@@ -171,7 +189,7 @@ class TombstoneCrashArtifact(AndroidArtifact):
# Validate the tombstone and add it to the results # Validate the tombstone and add it to the results
tombstone = TombstoneCrashResult.model_validate(tombstone_dict) tombstone = TombstoneCrashResult.model_validate(tombstone_dict)
self.results.append(tombstone.model_dump()) return tombstone.model_dump()
def _parse_tombstone_line( def _parse_tombstone_line(
self, line: str, key: str, destination_key: str, tombstone: dict self, line: str, key: str, destination_key: str, tombstone: dict
@@ -200,7 +218,9 @@ class TombstoneCrashArtifact(AndroidArtifact):
if not separator or line_key != key: if not separator or line_key != key:
return False 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": if destination_key == "uid":
tombstone[destination_key] = int(value_clean) tombstone[destination_key] = int(value_clean)
elif destination_key == "process_uptime": elif destination_key == "process_uptime":
@@ -274,9 +294,7 @@ class TombstoneCrashArtifact(AndroidArtifact):
@staticmethod @staticmethod
def _parse_timestamp_string(timestamp: str) -> str: def _parse_timestamp_string(timestamp: str) -> str:
timestamp_parsed = parser.parse(timestamp) timestamp_parsed = parser.parse(timestamp)
# Preserve the source wall-clock time while returning the project-wide ISO format. return convert_datetime_to_iso(timestamp_parsed)
local_timestamp = timestamp_parsed.replace(tzinfo=datetime.timezone.utc)
return convert_datetime_to_iso(local_timestamp)
@staticmethod @staticmethod
def _proccess_name_from_thread(tombstone_dict: dict) -> str: 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 # Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/ # https://license.mvt.re/1.1/
from mvt.common.module import MVTModule
from .aqf_files import AQFFiles from .aqf_files import AQFFiles
from .aqf_getprop import AQFGetProp from .aqf_getprop import AQFGetProp
from .aqf_log_timestamps import AQFLogTimestamps from .aqf_log_timestamps import AQFLogTimestamps
@@ -12,7 +14,7 @@ from .aqf_settings import AQFSettings
from .mounts import Mounts from .mounts import Mounts
from .root_binaries import RootBinaries from .root_binaries import RootBinaries
ANDROIDQF_MODULES = [ ANDROIDQF_MODULES: list[type[MVTModule]] = [
AQFPackages, AQFPackages,
AQFProcesses, AQFProcesses,
AQFGetProp, 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 # Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/ # https://license.mvt.re/1.1/
from mvt.common.module import MVTModule
from .sms import SMS 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 # Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/ # https://license.mvt.re/1.1/
from mvt.common.module import MVTModule
from .dumpsys_accessibility import DumpsysAccessibility from .dumpsys_accessibility import DumpsysAccessibility
from .dumpsys_activities import DumpsysActivities from .dumpsys_activities import DumpsysActivities
from .dumpsys_appops import DumpsysAppops from .dumpsys_appops import DumpsysAppops
@@ -16,8 +18,11 @@ from .dumpsys_receivers import DumpsysReceivers
from .dumpsys_adb_state import DumpsysADBState from .dumpsys_adb_state import DumpsysADBState
from .fs_timestamps import BugReportTimestamps from .fs_timestamps import BugReportTimestamps
from .tombstones import Tombstones from .tombstones import Tombstones
from .mounts import Mounts
from .processes import Processes
from .settings import Settings
BUGREPORT_MODULES = [ BUGREPORT_MODULES: list[type[MVTModule]] = [
DumpsysAccessibility, DumpsysAccessibility,
DumpsysActivities, DumpsysActivities,
DumpsysAppops, DumpsysAppops,
@@ -31,4 +36,7 @@ BUGREPORT_MODULES = [
DumpsysADBState, DumpsysADBState,
BugReportTimestamps, BugReportTimestamps,
Tombstones, Tombstones,
Processes,
Settings,
Mounts,
] ]
+26 -1
View File
@@ -72,7 +72,11 @@ class BugReportModule(MVTModule):
if not self.extract_path: if not self.extract_path:
raise ValueError("extract_path is not set") raise ValueError("extract_path is not set")
joined = os.path.join(self.extract_path, file_path) 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") raise ValueError("unsafe file_path")
handle = open(joined, "rb") handle = open(joined, "rb")
@@ -100,6 +104,27 @@ class BugReportModule(MVTModule):
return None 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: def _get_file_modification_time(self, file_path: str) -> datetime.datetime:
if self.zip_archive: if self.zip_archive:
file_timetuple = self.zip_archive.getinfo(file_path).date_time file_timetuple = self.zip_archive.getinfo(file_path).date_time
@@ -49,9 +49,7 @@ class DumpsysAccessibility(DumpsysAccessibilityArtifact, BugReportModule):
self.parse(content) self.parse(content)
for result in self.results: for result in self.results:
self.log.info( self.log.info('Found accessibility service "%s"', result.get("component"))
'Found installed accessibility service "%s"', result.get("service")
)
self.log.info( self.log.info(
"Identified a total of %d accessibility services", len(self.results) "Identified a total of %d accessibility services", len(self.results)
@@ -44,21 +44,8 @@ class DumpsysGetProp(GetPropArtifact, BugReportModule):
) )
return return
lines = [] section = self.extract_command_section(
in_getprop = False content.decode(errors="ignore"), "------ SYSTEM PROPERTIES"
)
for line in content.decode(errors="ignore").splitlines(): self.parse(section)
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))
self.log.info("Extracted %d Android system properties", len(self.results)) self.log.info("Extracted %d Android system properties", len(self.results))
@@ -50,7 +50,10 @@ class DumpsysPackages(DumpsysPackagesArtifact, BugReportModule):
for result in self.results: for result in self.results:
dangerous_permissions_count = 0 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: if perm["name"] in DANGEROUS_PERMISSIONS:
dangerous_permissions_count += 1 dangerous_permissions_count += 1
@@ -13,7 +13,7 @@ from mvt.common.module_types import ModuleResults
class DumpsysPlatformCompat(DumpsysPlatformCompatArtifact, BugReportModule): class DumpsysPlatformCompat(DumpsysPlatformCompatArtifact, BugReportModule):
"""This module extracts details on uninstalled apps.""" """This module extracts raw per-package compatibility overrides."""
def __init__( def __init__(
self, self,
@@ -48,4 +48,4 @@ class DumpsysPlatformCompat(DumpsysPlatformCompatArtifact, BugReportModule):
) )
self.parse(content) 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, results=results,
) )
self.results = results if results else {} self.results = results if results else []
def run(self) -> None: def run(self) -> None:
content = self._get_dumpstate_file() content = self._get_dumpstate_file()
@@ -49,4 +49,4 @@ class DumpsysReceivers(DumpsysReceiversArtifact, BugReportModule):
) )
self.parse(dumpsys_section) 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/ # https://license.mvt.re/1.1/
import logging import logging
import datetime
from typing import Optional from typing import Optional
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from mvt.common.utils import convert_datetime_to_iso from mvt.common.utils import convert_datetime_to_iso
from .base import BugReportModule from .base import BugReportModule
from mvt.common.module_types import ModuleResults from mvt.common.module_types import ModuleResults
from mvt.android.artifacts.file_timestamps import FileTimestampsArtifact from mvt.android.artifacts.file_timestamps import FileTimestampsArtifact
from mvt.android.artifacts.getprop import GetProp
class BugReportTimestamps(FileTimestampsArtifact, BugReportModule): class BugReportTimestamps(FileTimestampsArtifact, BugReportModule):
@@ -38,15 +41,44 @@ class BugReportTimestamps(FileTimestampsArtifact, BugReportModule):
def run(self) -> None: def run(self) -> None:
filesystem_files = self._get_files_by_pattern("FS/*") 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 = [] self.results = []
for file in filesystem_files: for file in filesystem_files:
# Only the modification time is available in the zip file metadata. # Only the modification time is available in the zip file metadata.
# The timezone is the local timezone of the machine the phone. # The timezone is the local timezone of the machine the phone.
modification_time = self._get_file_modification_time(file) 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( self.results.append(
{ {
"path": file, "path": file,
"modified_time": convert_datetime_to_iso(modification_time), "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 return
for tombstone_file in sorted(tombstone_files): grouped: dict[str, dict[str, str]] = {}
tombstone_filename = tombstone_file.split("/")[-1] for file_path in tombstone_files:
modification_time = self._get_file_modification_time(tombstone_file) file_name = file_path.rsplit("/", 1)[-1]
tombstone_data = self._get_file_content(tombstone_file) source = "protobuf" if file_name.endswith(".pb") else "text"
crash_id = file_name.removesuffix(".pb")
grouped.setdefault(crash_id, {})[source] = file_path
try: for crash_id, paths in sorted(grouped.items()):
if tombstone_file.endswith(".pb"): parsed_sources: dict[str, dict] = {}
self.parse_protobuf( source_records: dict[str, dict] = {}
tombstone_filename, modification_time, tombstone_data 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: parsed_sources[source] = source_info
self.parse(tombstone_filename, modification_time, tombstone_data)
except ValueError as e: if not source_records:
# Catch any exceptions raised during parsing or validation. continue
self.log.error(f"Error parsing tombstone file {tombstone_file}: {e}") 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( self.log.info(
"Extracted a total of %d tombstone files", "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 # Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/ # https://license.mvt.re/1.1/
from mvt.common.module import MVTModule
from .connect_event import ConnectEvent from .connect_event import ConnectEvent
from .dns_event import DnsEvent from .dns_event import DnsEvent
from .security_event import SecurityEvent from .security_event import SecurityEvent
INTRUSION_LOGS_MODULES = [ INTRUSION_LOGS_MODULES: list[type[MVTModule]] = [
DnsEvent, DnsEvent,
ConnectEvent, ConnectEvent,
SecurityEvent, SecurityEvent,
+1
View File
@@ -39,6 +39,7 @@ HELP_MSG_DECRYPT_BACKUP = "Decrypt an encrypted iTunes backup"
HELP_MSG_BACKUP_DESTINATION = ( HELP_MSG_BACKUP_DESTINATION = (
"Path to the folder where the decrypted backup should be stored" "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 = ( HELP_MSG_IOS_BACKUP_PASSWORD = (
"Password to use to decrypt the backup (or, set the {MVT_IOS_BACKUP_PASSWORD} " "Password to use to decrypt the backup (or, set the {MVT_IOS_BACKUP_PASSWORD} "
"environment variable)" "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 # Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/ # 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_VERSION,
HELP_MSG_DECRYPT_BACKUP, HELP_MSG_DECRYPT_BACKUP,
HELP_MSG_BACKUP_DESTINATION, HELP_MSG_BACKUP_DESTINATION,
HELP_MSG_DECRYPT_JOBS,
HELP_MSG_IOS_BACKUP_PASSWORD, HELP_MSG_IOS_BACKUP_PASSWORD,
HELP_MSG_BACKUP_KEYFILE, HELP_MSG_BACKUP_KEYFILE,
HELP_MSG_HASHES, HELP_MSG_HASHES,
@@ -45,6 +46,10 @@ from mvt.common.help import (
HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK, HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK,
) )
from mvt.common.password import prompt_password 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 # 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 # 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 "decrypt-backup", context_settings=CONTEXT_SETTINGS, help=HELP_MSG_DECRYPT_BACKUP
) )
@click.option("--destination", "-d", required=True, help=HELP_MSG_BACKUP_DESTINATION) @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( @click.option(
"--password", "--password",
"-p", "-p",
@@ -146,10 +158,10 @@ def version():
@click.option("--hashes", "-H", is_flag=True, help=HELP_MSG_HASHES) @click.option("--hashes", "-H", is_flag=True, help=HELP_MSG_HASHES)
@click.argument("BACKUP_PATH", type=click.Path(exists=True)) @click.argument("BACKUP_PATH", type=click.Path(exists=True))
@click.pass_context @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 from .decrypt import DecryptBackup
backup = DecryptBackup(backup_path, destination) backup = DecryptBackup(backup_path, destination, max_workers=jobs)
if key_file: if key_file:
if MVT_IOS_BACKUP_PASSWORD in os.environ: if MVT_IOS_BACKUP_PASSWORD in os.environ:
+294 -76
View File
@@ -6,18 +6,179 @@
import binascii import binascii
import glob import glob
import logging import logging
import multiprocessing
import os import os
import os.path import os.path
import plistlib
import shutil import shutil
import sqlite3 import sqlite3
import tempfile
from concurrent.futures import (
ALL_COMPLETED,
FIRST_COMPLETED,
Future,
ThreadPoolExecutor,
wait,
)
from pathlib import Path from pathlib import Path
from typing import Optional 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__) 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: class DecryptBackup:
"""This class provides functions to decrypt an encrypted iTunes backup """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. """Decrypts an encrypted iOS backup.
:param backup_path: Path to the encrypted backup folder :param backup_path: Path to the encrypted backup folder
:param dest_path: Path to the folder where to store the decrypted backup :param dest_path: Path to the folder where to store the decrypted backup
""" """
self.backup_path = os.path.abspath(backup_path) self.backup_path = os.path.abspath(backup_path)
self.dest_path = dest_path self.dest_path = dest_path
self._backup = None if not 1 <= max_workers <= MAX_DECRYPT_WORKERS:
self._decryption_key = None 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: def can_process(self) -> bool:
return self._backup is not None 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 @staticmethod
def is_encrypted(backup_path: str) -> bool: def is_encrypted(backup_path: str) -> bool:
"""Query Manifest.db file to see if it's encrypted or not. """Query Manifest.db file to see if it's encrypted or not.
@@ -58,21 +268,6 @@ class DecryptBackup:
finally: finally:
conn.close() 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: def process_backup(self) -> None:
assert self._backup is not None assert self._backup is not None
assert self.dest_path is not None assert self.dest_path is not None
@@ -81,53 +276,71 @@ class DecryptBackup:
os.makedirs(self.dest_path) os.makedirs(self.dest_path)
manifest_path = os.path.join(self.dest_path, "Manifest.db") manifest_path = os.path.join(self.dest_path, "Manifest.db")
# We extract a decrypted Manifest.db. # Extract a decrypted Manifest.db to the destination folder.
self._backup.getManifestDB() self._backup.save_manifest_file(output_filename=manifest_path)
# We store it to the destination folder.
shutil.copy(self._backup.manifestDB, manifest_path)
pool = multiprocessing.Pool(multiprocessing.cpu_count()) # Iterate over all files in the backup and decrypt them,
# preserving the XX/file_id directory structure that downstream
for item in self._backup.getBackupFilesList(): # modules expect.
try: backup_root = Path(self.backup_path).resolve()
file_id = item["backupFile"] dest_root = Path(self.dest_path).resolve()
relative_path = item["relativePath"] pending: dict[Future[None], str] = {}
domain = item["domain"] with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
with self._backup.manifest_db_cursor() as cur:
# This may be a partial backup. Skip files from the manifest cur.execute(
# which do not exist locally. "SELECT fileID, domain, relativePath, file FROM Files WHERE flags=1"
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),
) )
except Exception as exc: for file_id, domain, relative_path, file_bplist in cur:
log.error("Failed to decrypt file %s: %s", relative_path, exc) # 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() output_path = dest_root / file_id[:2] / file_id
pool.join() 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. # Copying over the root plist files as well.
for file_name in os.listdir(self.backup_path): for file_name in os.listdir(self.backup_path):
@@ -168,20 +381,23 @@ class DecryptBackup:
return return
try: try:
self._backup = iOSbackup( self._backup = MVTEncryptedBackup(
udid=os.path.basename(self.backup_path), backup_directory=self.backup_path,
cleartextpassword=password, passphrase=password,
backuproot=os.path.dirname(self.backup_path),
) )
# Eagerly trigger keybag unlock so wrong-password errors
# surface here rather than later during process_backup().
self._backup.test_decryption()
except Exception as exc: except Exception as exc:
self._backup = None
if ( if (
isinstance(exc, KeyError) isinstance(exc, ValueError)
and len(exc.args) > 0 and "passphrase" in str(exc).lower()
and exc.args[0] == b"KEY"
): ):
log.critical("Failed to decrypt backup. Password is probably wrong.") log.critical("Failed to decrypt backup. Password is probably wrong.")
elif ( elif (
isinstance(exc, FileNotFoundError) isinstance(exc, FileNotFoundError)
and hasattr(exc, "filename")
and os.path.basename(exc.filename) == "Manifest.plist" and os.path.basename(exc.filename) == "Manifest.plist"
): ):
log.critical( log.critical(
@@ -224,12 +440,14 @@ class DecryptBackup:
try: try:
key_bytes_raw = binascii.unhexlify(key_bytes) key_bytes_raw = binascii.unhexlify(key_bytes)
self._backup = iOSbackup( self._backup = MVTEncryptedBackup(
udid=os.path.basename(self.backup_path), backup_directory=self.backup_path,
derivedkey=key_bytes_raw, derived_key=key_bytes_raw,
backuproot=os.path.dirname(self.backup_path),
) )
# Eagerly trigger keybag unlock so wrong-key errors surface here.
self._backup.test_decryption()
except Exception as exc: except Exception as exc:
self._backup = None
log.exception(exc) log.exception(exc)
log.critical( log.critical(
"Failed to decrypt backup. Did you provide the correct key file?" "Failed to decrypt backup. Did you provide the correct key file?"
@@ -240,7 +458,7 @@ class DecryptBackup:
if not self._backup: if not self._backup:
return return
self._decryption_key = self._backup.getDecryptionKey() self._decryption_key = self._backup.get_decryption_key()
log.info( log.info(
'Derived decryption key for backup at path %s is: "%s"', 'Derived decryption key for backup at path %s is: "%s"',
self.backup_path, 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 # Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/ # https://license.mvt.re/1.1/
from mvt.common.module import MVTModule
from .backup_info import BackupInfo from .backup_info import BackupInfo
from .configuration_profiles import ConfigurationProfiles from .configuration_profiles import ConfigurationProfiles
from .manifest import Manifest from .manifest import Manifest
from .profile_events import ProfileEvents 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 # Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/ # https://license.mvt.re/1.1/
from mvt.common.module import MVTModule
from .analytics import Analytics from .analytics import Analytics
from .analytics_ios_versions import AnalyticsIOSVersions from .analytics_ios_versions import AnalyticsIOSVersions
from .cache_files import CacheFiles from .cache_files import CacheFiles
@@ -15,7 +17,7 @@ from .webkit_indexeddb import WebkitIndexedDB
from .webkit_localstorage import WebkitLocalStorage from .webkit_localstorage import WebkitLocalStorage
from .webkit_safariviewservice import WebkitSafariViewService from .webkit_safariviewservice import WebkitSafariViewService
FS_MODULES = [ FS_MODULES: list[type[MVTModule]] = [
CacheFiles, CacheFiles,
Filesystem, Filesystem,
Netusage, 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 # Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/ # https://license.mvt.re/1.1/
from mvt.common.module import MVTModule
from .applications import Applications from .applications import Applications
from .calendar import Calendar from .calendar import Calendar
from .calls import Calls from .calls import Calls
@@ -28,7 +30,7 @@ from .webkit_session_resource_log import WebkitSessionResourceLog
from .whatsapp import Whatsapp from .whatsapp import Whatsapp
from .whatsapp_contacts import WhatsappContacts from .whatsapp_contacts import WhatsappContacts
MIXED_MODULES = [ MIXED_MODULES: list[type[MVTModule]] = [
Calls, Calls,
ChromeFavicon, ChromeFavicon,
ChromeHistory, ChromeHistory,
@@ -23,7 +23,7 @@ class TestDumpsysAccessibilityArtifact:
assert len(da.results) == 4 assert len(da.results) == 4
assert da.results[0]["package_name"] == "com.android.settings" assert da.results[0]["package_name"] == "com.android.settings"
assert ( assert (
da.results[0]["service"] da.results[0]["component"]
== "com.android.settings/com.samsung.android.settings.development.gpuwatch.GPUWatchInterceptor" == "com.android.settings/com.samsung.android.settings.development.gpuwatch.GPUWatchInterceptor"
) )
@@ -37,7 +37,9 @@ class TestDumpsysAccessibilityArtifact:
da.parse(data) da.parse(data)
assert len(da.results) == 1 assert len(da.results) == 1
assert da.results[0]["package_name"] == "com.malware.accessibility" 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): def test_accessibility_service_alert(self):
da = DumpsysAccessibilityArtifact() da = DumpsysAccessibilityArtifact()
@@ -52,6 +54,22 @@ class TestDumpsysAccessibilityArtifact:
assert da.alertstore.alerts[0].level == AlertLevel.MEDIUM assert da.alertstore.alerts[0].level == AlertLevel.MEDIUM
assert da.alertstore.alerts[0].event == da.results[0] 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): def test_ioc_check(self, indicator_file):
da = DumpsysAccessibilityArtifact() da = DumpsysAccessibilityArtifact()
file = get_artifact("android_data/dumpsys_accessibility.txt") 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 # Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/ # https://license.mvt.re/1.1/
import base64
from mvt.android.artifacts.dumpsys_adb import DumpsysADBArtifact from mvt.android.artifacts.dumpsys_adb import DumpsysADBArtifact
from mvt.android.modules.bugreport.dumpsys_adb_state import DumpsysADBState from mvt.android.modules.bugreport.dumpsys_adb_state import DumpsysADBState
from mvt.common.alerts import AlertLevel from mvt.common.alerts import AlertLevel
@@ -11,6 +13,17 @@ from ..utils import get_artifact
class TestDumpsysADBArtifact: 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): def test_parsing(self):
da_adb = DumpsysADBArtifact() da_adb = DumpsysADBArtifact()
file = get_artifact("android_data/dumpsys_adb.txt") 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 da.results[0]["uid"] == "0"
assert len(da.results[0]["permissions"]) == 1 assert len(da.results[0]["permissions"]) == 1
assert da.results[0]["permissions"][0]["name"] == "MANAGE_IPSEC_TUNNELS" 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 da.results[6]["package_name"] == "com.sec.factory.camera"
assert len(da.results[6]["permissions"][1]["entries"]) == 1 assert len(da.results[6]["permissions"][1]["entries"]) == 1
assert len(da.results[11]["permissions"]) == 4 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): def test_ioc_check(self, indicator_file):
da = DumpsysAppopsArtifact() da = DumpsysAppopsArtifact()
@@ -57,18 +57,19 @@ class TestDumpsysBatteryDailyArtifact:
assert uninstall_alert.message == ( assert uninstall_alert.message == (
"Detected uninstall of package com.example.removed (vers 0)" "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["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.level == AlertLevel.MEDIUM
assert downgrade_alert.message == ( assert downgrade_alert.message == (
"Detected downgrade of package com.example.app from vers 10 to vers 9" "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["package_name"] == "com.example.app"
assert downgrade_alert.event["action"] == "downgrade" 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): def test_newest_first_update_is_not_reported_as_downgrade(self):
dba = DumpsysBatteryDailyArtifact() dba = DumpsysBatteryDailyArtifact()
@@ -107,7 +108,19 @@ class TestDumpsysBatteryDailyArtifact:
assert downgrade_alert.event_time == "2026-01-10" assert downgrade_alert.event_time == "2026-01-10"
assert downgrade_alert.event["package_name"] == "com.example.app" assert downgrade_alert.event["package_name"] == "com.example.app"
assert downgrade_alert.event["action"] == "downgrade" 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): def test_reinstall_after_uninstall_is_not_reported_as_downgrade(self):
dba = DumpsysBatteryDailyArtifact() dba = DumpsysBatteryDailyArtifact()
@@ -54,6 +54,7 @@ class TestDumpsysBatteryHistoryArtifact:
assert len(dba.results) == 2 assert len(dba.results) == 2
assert dba.results[0] == { assert dba.results[0] == {
"time_elapsed": "07-15 20:27:39.431", "time_elapsed": "07-15 20:27:39.431",
"timestamp": "1900-07-15 20:27:39.431000",
"event": "start_job", "event": "start_job",
"uid": "u0a123", "uid": "u0a123",
"package_name": "com.example", "package_name": "com.example",
@@ -61,3 +62,23 @@ class TestDumpsysBatteryHistoryArtifact:
} }
assert dba.results[1]["event"] == "end_job" assert dba.results[1]["event"] == "end_job"
assert dba.results[1]["uid"] == "u0a123" 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 == [ assert dbi.results == [
{ {
"isodate": "07-15 20:27:39.431", "timestamp": "07-15 20:27:39.431",
"pid": None,
"action": "executeForCursorWindow", "action": "executeForCursorWindow",
"duration_ms": 1,
"status": "succeeded",
"sql": "SELECT 1", "sql": "SELECT 1",
"path": "/data/user/0/com.example/databases/current.db", "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 assert len(dpa.results) == 0
dpa.parse(data) dpa.parse(data)
assert len(dpa.results) == 4 assert len(dpa.results) == 10
assert dpa.results[0]["package_name"] == "com.samsung.android.app.social" assert dpa.results[0]["package_name"] == "com.samsung.android.messaging"
assert ( assert (
dpa.results[0]["activity"] dpa.results[0]["component"]
== "com.samsung.android.app.social/.feed.FeedsActivity" == "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): def test_ioc_check(self, indicator_file):
dpa = DumpsysPackageActivitiesArtifact() dpa = DumpsysPackageActivitiesArtifact()
@@ -41,4 +45,4 @@ class TestDumpsysPackageActivitiesArtifact:
dpa.indicators = ind dpa.indicators = ind
assert len(dpa.alertstore.alerts) == 0 assert len(dpa.alertstore.alerts) == 0
dpa.check_indicators() 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" == "com.samsung.android.provider.filterprovider"
) )
assert dpa.results[0]["version_name"] == "5.0.07" 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 assert dpa.results[0]["system"] is True
def test_parsing_system_flag(self): def test_parsing_system_flag(self):
@@ -60,42 +64,27 @@ class TestDumpsysPackagesArtifact:
dpa.check_indicators() dpa.check_indicators()
assert len(dpa.alertstore.alerts) == 1 assert len(dpa.alertstore.alerts) == 1
def test_per_user_fields_use_primary_user(self): def test_hidden_packages_and_per_user_state(self):
details = DumpsysPackagesArtifact.parse_dumpsys_package_for_details( dpa = DumpsysPackagesArtifact()
""" User 0: installed=true dpa.parse(
firstInstallTime=2024-01-10 09:19:39 """Packages:
runtime permissions: Package [com.example.active]:
android.permission.CAMERA: granted=true appId=10001
User 95: installed=false versionCode=12 minSdk=29 targetSdk=35
firstInstallTime=1970-01-01 01:00:00 User 0: installed=true hidden=false
runtime permissions: firstInstallTime=2025-01-01 01:02:03
android.permission.CAMERA: granted=false User 10: installed=false hidden=true
android.permission.RECORD_AUDIO: granted=false 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" assert [record["package_type"] for record in dpa.results] == [
runtime_permissions = [ "active",
permission "hidden_system",
for permission in details["permissions"]
if permission["type"] == "runtime"
] ]
assert runtime_permissions == [ assert len(dpa.results[0]["users"]) == 2
{ assert dpa.results[1]["installer"] is None
"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"
@@ -22,6 +22,9 @@ class TestDumpsysPlatformCompatArtifact:
assert len(dbi.results) == 2 assert len(dbi.results) == 2
assert dbi.results[0]["package_name"] == "org.torproject.torbrowser" assert dbi.results[0]["package_name"] == "org.torproject.torbrowser"
assert dbi.results[1]["package_name"] == "org.article19.circulo.next" 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): def test_ioc_check(self, indicator_file):
dbi = DumpsysPlatformCompatArtifact() dbi = DumpsysPlatformCompatArtifact()
@@ -19,17 +19,14 @@ class TestDumpsysReceiversArtifact:
assert len(dr.results) == 0 assert len(dr.results) == 0
dr.parse(data) dr.parse(data)
assert len(dr.results) == 4 assert len(dr.results) == 9
assert ( assert dr.results[0]["resolver_type"] == "full_mime_type"
list(dr.results.keys())[0] storage_manager = next(
== "com.android.storagemanager.automatic.SHOW_NOTIFICATION" result
) for result in dr.results
assert ( if result["key"] == "com.android.storagemanager.automatic.SHOW_NOTIFICATION"
dr.results["com.android.storagemanager.automatic.SHOW_NOTIFICATION"][0][
"package_name"
]
== "com.android.storagemanager"
) )
assert storage_manager["package_name"] == "com.android.storagemanager"
def test_parsing_misindented_action(self): def test_parsing_misindented_action(self):
dr = DumpsysReceiversArtifact() dr = DumpsysReceiversArtifact()
@@ -44,12 +41,8 @@ Receiver Resolver Table:
dr.parse(data) dr.parse(data)
assert ( assert dr.results[1]["key"] == "android.intent.action.MY_PACKAGE_REPLACED"
dr.results["android.intent.action.MY_PACKAGE_REPLACED"][0][ assert dr.results[1]["package_name"] == "com.psycatgames.nhiegame"
"package_name"
]
== "com.psycatgames.nhiegame"
)
def test_parsing_misindented_first_action(self): def test_parsing_misindented_first_action(self):
dr = DumpsysReceiversArtifact() dr = DumpsysReceiversArtifact()
@@ -62,12 +55,8 @@ Receiver Resolver Table:
dr.parse(data) dr.parse(data)
assert ( assert dr.results[0]["key"] == "android.intent.action.MY_PACKAGE_REPLACED"
dr.results["android.intent.action.MY_PACKAGE_REPLACED"][0][ assert dr.results[0]["package_name"] == "com.psycatgames.nhiegame"
"package_name"
]
== "com.psycatgames.nhiegame"
)
def test_ioc_check(self, indicator_file): def test_ioc_check(self, indicator_file):
dr = DumpsysReceiversArtifact() dr = DumpsysReceiversArtifact()
+26
View File
@@ -39,3 +39,29 @@ class TestGetPropArtifact:
assert len(gp.alertstore.alerts) == 0 assert len(gp.alertstore.alerts) == 0
gp.check_indicators() gp.check_indicators()
assert len(gp.alertstore.alerts) == 1 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 assert len(p.results) == 0
p.parse(data) p.parse(data)
assert len(p.results) == 17 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): def test_ioc_check(self, indicator_file):
p = Processes() p = Processes()
@@ -36,3 +36,14 @@ class TestProcessesArtifact:
assert len(p.alertstore.alerts) == 0 assert len(p.alertstore.alerts) == 0
p.check_indicators() p.check_indicators()
assert len(p.alertstore.alerts) == 1 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("pid") == 25541
assert tombstone_result.get("process_name") == "mtk.ape.decoder" assert tombstone_result.get("process_name") == "mtk.ape.decoder"
# With Android logs we want to keep timestamps as device local time for consistency. # Tombstones include an explicit offset, so normalize them to UTC.
# We often don't know the time offset for a log entry and so can't convert everything to UTC. assert tombstone_result.get("timestamp") == "2023-04-12 10:32:40.518290"
# 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"
+13
View File
@@ -13,6 +13,19 @@ from ..utils import get_android_androidqf, list_files
class TestAndroidqfMountsArtifact: 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): def test_parse_mounts_token_checks(self):
""" """
Test the artifact-level `parse` method using tolerant token checks. Test the artifact-level `parse` method using tolerant token checks.
+15
View File
@@ -6,12 +6,27 @@
from pathlib import Path from pathlib import Path
from mvt.android.modules.androidqf.aqf_settings import AQFSettings from mvt.android.modules.androidqf.aqf_settings import AQFSettings
from mvt.android.artifacts.settings import Settings
from mvt.common.module import run_module from mvt.common.module import run_module
from ..utils import get_android_androidqf, list_files from ..utils import get_android_androidqf, list_files
class TestSettingsModule: 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): def test_parsing(self):
data_path = get_android_androidqf() data_path = get_android_androidqf()
m = AQFSettings(target_path=data_path) m = AQFSettings(target_path=data_path)
+14 -8
View File
@@ -54,10 +54,11 @@ class TestBugreportAnalysis:
== "com.samsung.android.provider.filterprovider" == "com.samsung.android.provider.filterprovider"
) )
assert m.results[1]["package_name"] == "com.instagram.android" 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 m.results[1]["installer"] == "com.android.vending"
assert len(m.results[0]["permissions"]) == 4 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): def test_getprop_module(self):
m = self.launch_bug_report_module(DumpsysGetProp) m = self.launch_bug_report_module(DumpsysGetProp)
@@ -66,29 +67,34 @@ class TestBugreportAnalysis:
def test_receivers_match_exact_package_name(self, indicators_factory): def test_receivers_match_exact_package_name(self, indicators_factory):
intent = "android.intent.action.PHONE_STATE" intent = "android.intent.action.PHONE_STATE"
false_positive = { false_positive = {
"resolver_type": "non_data_action",
"key": intent,
"package_name": "com.android.phone", "package_name": "com.android.phone",
"receiver": ( "component": (
"com.android.phone/" "com.android.phone/"
"com.android.services.telephony.sip.SipIncomingCallReceiver" "com.android.services.telephony.sip.SipIncomingCallReceiver"
), ),
"filter_count": 1,
} }
malicious_receiver = { malicious_receiver = {
"resolver_type": "non_data_action",
"key": intent,
"package_name": "com.android.services", "package_name": "com.android.services",
"receiver": "com.android.services/com.example.SomeReceiver", "component": "com.android.services/com.example.SomeReceiver",
"filter_count": 1,
} }
module = DumpsysReceivers( module = DumpsysReceivers(results=[false_positive, malicious_receiver])
results={intent: [false_positive, malicious_receiver]}
)
module.indicators = indicators_factory(app_ids=["com.android.services"]) module.indicators = indicators_factory(app_ids=["com.android.services"])
module.check_indicators() module.check_indicators()
assert len(module.alertstore.alerts) == 1 assert len(module.alertstore.alerts) == 1
alert = module.alertstore.alerts[0] alert = module.alertstore.alerts[0]
assert alert.event == {intent: malicious_receiver} assert alert.event == malicious_receiver
assert alert.matched_indicator.value == "com.android.services" assert alert.matched_indicator.value == "com.android.services"
def test_tombstones_modules(self): def test_tombstones_modules(self):
m = self.launch_bug_report_module(Tombstones) m = self.launch_bug_report_module(Tombstones)
assert len(m.results) == 2 assert len(m.results) == 2
assert m.results[1]["pid"] == 3559 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]] [[package]]
name = "iosbackup" name = "iphone-backup-decrypt"
version = "0.9.925" version = "0.9.0"
source = { registry = "https://pypi.org/simple" } source = { registry = "https://pypi.org/simple" }
dependencies = [ dependencies = [
{ name = "nskeyedunarchiver" },
{ name = "pycryptodome" }, { 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 = [ 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]] [[package]]
@@ -1022,11 +1021,12 @@ dependencies = [
{ name = "betterproto2" }, { name = "betterproto2" },
{ name = "click" }, { name = "click" },
{ name = "cryptography" }, { name = "cryptography" },
{ name = "iosbackup" }, { name = "iphone-backup-decrypt" },
{ name = "libusb1" }, { name = "libusb1" },
{ name = "nskeyedunarchiver" }, { name = "nskeyedunarchiver" },
{ name = "packaging" }, { name = "packaging" },
{ name = "pyahocorasick" }, { name = "pyahocorasick" },
{ name = "pycryptodome" },
{ name = "pydantic" }, { name = "pydantic" },
{ name = "pydantic-settings" }, { name = "pydantic-settings" },
{ name = "python-dateutil" }, { name = "python-dateutil" },
@@ -1068,11 +1068,12 @@ requires-dist = [
{ name = "betterproto2", specifier = "==0.10.0" }, { name = "betterproto2", specifier = "==0.10.0" },
{ name = "click", specifier = "==8.4.2" }, { name = "click", specifier = "==8.4.2" },
{ name = "cryptography", specifier = "==50.0.0" }, { 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 = "libusb1", specifier = "==3.4.0" },
{ name = "nskeyedunarchiver", specifier = "==1.5.2" }, { name = "nskeyedunarchiver", specifier = "==1.5.2" },
{ name = "packaging", specifier = "==26.3" }, { name = "packaging", specifier = "==26.3" },
{ name = "pyahocorasick", specifier = "==2.3.1" }, { name = "pyahocorasick", specifier = "==2.3.1" },
{ name = "pycryptodome", specifier = ">=3.20.0" },
{ name = "pydantic", specifier = "==2.13.4" }, { name = "pydantic", specifier = "==2.13.4" },
{ name = "pydantic-settings", specifier = "==2.15.0" }, { name = "pydantic-settings", specifier = "==2.15.0" },
{ name = "python-dateutil", specifier = "==2.9.0.post0" }, { name = "python-dateutil", specifier = "==2.9.0.post0" },