Compare commits

..
68 changed files with 734 additions and 2040 deletions

No files matched your search

+9 -88
View File
@@ -14,10 +14,10 @@ env:
REGISTRY: ghcr.io
IMAGE_NAME: ${{ github.repository }}
# Build each architecture natively, then publish the combined image manifests.
# There is a single job in this workflow. It's configured to run on the latest available version of Ubuntu.
jobs:
build-and-push-image:
runs-on: ${{ matrix.architecture.runner }}
runs-on: ubuntu-latest
# Sets the permissions granted to the `GITHUB_TOKEN` for the actions in this job.
permissions:
contents: read
@@ -26,15 +26,7 @@ jobs:
id-token: write
#
strategy:
fail-fast: false
matrix:
architecture:
- platform: linux/amd64
runner: ubuntu-24.04
name: amd64
- platform: linux/arm64
runner: ubuntu-24.04-arm
name: arm64
platform:
- dockerfile: "Dockerfile"
tag-suffix: ""
@@ -69,95 +61,24 @@ jobs:
type=raw,enable=${{ github.ref_type == 'tag' }},value=${{ github.ref_name }},suffix=${{ matrix.platform.tag-suffix }}
type=sha,suffix=${{ matrix.platform.tag-suffix }}
type=sha,format=long,suffix=${{ matrix.platform.tag-suffix }}
# Register emulators so Buildx can build the ARM64 images on the AMD64 runner.
- name: Set up QEMU
uses: docker/setup-qemu-action@v4
# This step sets up some additional capabilities to generate the provenance and sbom attestations
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v4
# This step uses the `docker/build-push-action` action to build the image, based on your repository's `Dockerfile`. If the build succeeds, it pushes the image to GitHub Packages.
# It uses the `context` parameter to define the build's context as the set of files located in the specified path. For more information, see "[Usage](https://github.com/docker/build-push-action#usage)" in the README of the `docker/build-push-action` repository.
# Push by digest so architecture builds do not overwrite shared image tags.
# It uses the `tags` and `labels` parameters to tag and label the image with the output from the "meta" step.
- name: Build and push Docker image
id: push
uses: docker/build-push-action@v7
with:
file: ${{ matrix.platform.dockerfile }}
context: .
platforms: ${{ matrix.architecture.platform }}
outputs: type=image,name=${{ env.REGISTRY }}/${{ env.IMAGE_NAME }},push-by-digest=true,name-canonical=true,push=true
platforms: linux/amd64,linux/arm64
push: true
tags: ${{ steps.meta.outputs.tags }}
labels: ${{ steps.meta.outputs.labels }}
provenance: mode=max
sbom: true
- name: Export image digest
env:
DIGEST: ${{ steps.push.outputs.digest }}
run: |
mkdir -p "${RUNNER_TEMP}/digests"
touch "${RUNNER_TEMP}/digests/${DIGEST#sha256:}"
- name: Upload image digest
uses: actions/upload-artifact@v4
with:
name: digests-${{ matrix.platform.dockerfile }}-${{ matrix.architecture.name }}
path: ${{ runner.temp }}/digests/*
if-no-files-found: error
retention-days: 1
publish-manifests:
needs: build-and-push-image
runs-on: ubuntu-24.04
permissions:
contents: read
packages: write
strategy:
matrix:
platform:
- dockerfile: "Dockerfile"
tag-suffix: ""
- dockerfile: "Dockerfile.ios"
tag-suffix: "-ios"
- dockerfile: "Dockerfile.android"
tag-suffix: "-android"
steps:
- name: Download image digests
uses: actions/download-artifact@v4
with:
pattern: digests-${{ matrix.platform.dockerfile }}-*
path: ${{ runner.temp }}/digests
merge-multiple: true
- name: Log in to the Container registry
uses: docker/login-action@dbcb813823bdd20940b903addbd779551569679f # v4.6.0
with:
registry: ${{ env.REGISTRY }}
username: ${{ github.actor }}
password: ${{ secrets.GITHUB_TOKEN }}
- name: Extract metadata (tags, labels) for Docker
id: meta
uses: docker/metadata-action@v6
with:
images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
flavor: |
latest=false
tags: |
type=raw,value=latest,enable={{ is_default_branch }},suffix=${{ matrix.platform.tag-suffix }}
type=raw,enable=${{ github.event_name == 'release' || github.ref_type == 'tag' }},value=stable,suffix=${{ matrix.platform.tag-suffix }}
type=raw,enable=${{ github.event_name == 'release' }},value=${{ github.event.release.tag_name }},suffix=${{ matrix.platform.tag-suffix }}
type=raw,enable=${{ github.ref_type == 'tag' }},value=${{ github.ref_name }},suffix=${{ matrix.platform.tag-suffix }}
type=sha,suffix=${{ matrix.platform.tag-suffix }}
type=sha,format=long,suffix=${{ matrix.platform.tag-suffix }}
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v4
- name: Publish multi-platform image
working-directory: ${{ runner.temp }}/digests
env:
METADATA_JSON: ${{ steps.meta.outputs.json }}
run: |
mapfile -t tags < <(jq -r '.tags[]' <<< "$METADATA_JSON")
tag_args=()
for tag in "${tags[@]}"; do
tag_args+=(--tag "$tag")
done
images=()
for digest in *; do
images+=("${REGISTRY}/${IMAGE_NAME}@sha256:${digest}")
done
test "${#images[@]}" -eq 2
docker buildx imagetools create "${tag_args[@]}" "${images[@]}"
docker buildx imagetools inspect "${tags[0]}"
-1
View File
@@ -34,7 +34,6 @@ jobs:
publish:
if: github.ref_type == 'tag'
runs-on: ubuntu-latest
environment: pypi # only tag refs may deploy; the PyPI trusted publisher requires it
permissions:
id-token: write # PyPI trusted publishing
steps:
-4
View File
@@ -1,4 +0,0 @@
prune tests
prune docs
prune .github
exclude uv.lock Dockerfile* mkdocs.yml .readthedocs.yaml .safety-policy.yml .python-version .gitignore Makefile
-8
View File
@@ -435,11 +435,3 @@ This JSON file is created by mvt-ios' `WhatsappContacts` module. The module extr
This database is often missing from incremental backups. When it cannot be found, the module logs a warning and produces no results, in which case the disappearing messages state of chats cannot be determined from the backup.
---
## Records extracted by `check-sysdiagnose`
### `sysdiagnose_info.json`
This JSON file is created by mvt-ios' `SysdiagnoseInfo` module. The module extracts details about the device and the sysdiagnose itself: the product type and model, iOS version and build, serial number, IMEI, MEID and UDID from the remotectl dump state and the mobile activation request, the Apple account name and email from the App Store daemon database (no longer part of a sysdiagnose on newer iOS versions, still read from older archives), and the original file name and creation time of the archive from *sysdiagnose.log*.
+4 -7
View File
@@ -1,13 +1,10 @@
# Check an iOS Sysdiagnose
`mvt-ios check-sysdiagnose` analyzes an iOS sysdiagnose archive. MVT's own
`SysdiagnoseInfo` module extracts details about the device and the archive
(see [`sysdiagnose_info.json`](records.md#sysdiagnose_infojson)); the checks
come from the modules of the installed
`mvt-ios check-sysdiagnose` prepares an iOS sysdiagnose archive for analysis by
custom MVT modules. MVT does not include built-in sysdiagnose modules. The
command runs the modules of the installed
[plugin packages](../development/index.md#installed-module-packages) which
declare support for the command. Without any such module the command still
records the device details, and warns that no forensic sysdiagnose modules
have been loaded so that the run cannot pass for a clean analysis.
declare support for it. Install at least one such package first.
The command accepts either an extracted sysdiagnose directory or the original
gzip-compressed tar archive.
+3 -1
View File
@@ -24,8 +24,10 @@ dependencies = [
"simplejson==4.1.1",
"packaging==26.3",
"appdirs==1.4.4",
"iphone_backup_decrypt==0.10.0",
"iphone_backup_decrypt==0.9.0",
"pycryptodome>=3.20.0",
"adb-shell[usb]==0.4.4",
"libusb1==3.4.0",
"cryptography==50.0.0",
"PyYAML>=6.0.2",
"pyahocorasick==2.3.1",
@@ -10,45 +10,8 @@ from .artifact import AndroidArtifact
class DumpsysAccessibilityArtifact(AndroidArtifact):
# One list for both record shapes — a service record and a count-only
# record must stay the same shape.
_FIELDS = (
"user_id",
"component",
"package_name",
"service_name",
"installed",
"enabled",
"binding",
"bound",
"crashed",
"accessibility_tool",
"installed_service_count",
"unnamed_service_count",
)
def check_indicators(self) -> None:
for result in self.results:
# A stated count the dump does not back with component names is a
# coverage statement, not a service: low, but not silent.
if not result.get("component"):
stated = result["installed_service_count"]
unnamed = result["unnamed_service_count"]
if unnamed == stated:
detail = "does not list their component names"
else:
detail = (
f"lists the component names of only {stated - unnamed} "
f"of them ({unnamed} unnamed)"
)
self.alertstore.low(
f"The accessibility dump states {stated} installed "
f"service(s) for user {result['user_id']} but {detail}",
"",
result,
)
continue
if self.indicators:
ioc_match = self.indicators.check_app_id(result["package_name"])
if ioc_match:
@@ -57,22 +20,11 @@ class DumpsysAccessibilityArtifact(AndroidArtifact):
)
continue
# Installed is not enabled. A service can sit installed for years
# without ever being switched on, and that is the difference this
# channel is read for — an alert that says only "found" makes every
# device look equally exposed. A service the dump says is switched
# OFF is reported LOW, so it still reaches the analyst without
# competing with one that is actually running; a dump that does not
# state the enabled state stays MEDIUM, because "not stated" is not
# "not enabled".
message = (
f'Found accessibility service: "{result["component"]}" '
f"({self._describe_state(result)})"
self.alertstore.medium(
f'Found accessibility service: "{result["component"]}"',
"",
result,
)
if result.get("enabled") is False and not result.get("bound"):
self.alertstore.low(message, "", result)
else:
self.alertstore.medium(message, "", result)
def parse(self, content: str) -> None:
"""
@@ -84,13 +36,6 @@ class DumpsysAccessibilityArtifact(AndroidArtifact):
self.results: list[dict[str, Any]] = []
services: dict[tuple[int | None, str], dict] = {}
# Which state sections the dump printed, per user: one user's printed
# `enabled services` says nothing about another user's.
seen_states: dict[int | None, set[str]] = {}
# `installedServiceCount=N` from the user's `attributes:{…}` line is on
# most builds the only statement about installed services in the dump:
# few print the `installed services: {…}` block.
installed_counts: dict[int | None, int] = {}
user_id: int | None = None
state: str | None = None
@@ -99,10 +44,6 @@ class DumpsysAccessibilityArtifact(AndroidArtifact):
if user_match:
user_id = int(user_match.group(1))
count_match = re.search(r"installedServiceCount=(\d+)", line)
if count_match:
installed_counts[user_id] = int(count_match.group(1))
stripped = line.strip()
state_match = re.match(
r"(?i)(installed|enabled|binding|bound|crashed) services\s*:\s*\{(.*)",
@@ -110,7 +51,6 @@ class DumpsysAccessibilityArtifact(AndroidArtifact):
)
if state_match:
state = state_match.group(1).lower()
seen_states.setdefault(user_id, set()).add(self._state_field(state))
inline = state_match.group(2)
for component in re.findall(
r"\{?([\w.$-]+/[\w.$-]+)(?:\s+\(A11yTool\))?\}?", inline
@@ -139,65 +79,24 @@ class DumpsysAccessibilityArtifact(AndroidArtifact):
service[self._state_field(state)] = True
service["accessibility_tool"] = "(A11yTool)" in stripped
# A section that was never printed is NOT the same as one printed
# empty: the first says nothing, the second says nothing is enabled.
# Defaulting every flag to False would turn "not stated" into "not
# enabled". Flags for sections this dump never printed stay None.
named: dict[int | None, int] = {}
for (service_user, _component), service in services.items():
printed = seen_states.get(service_user, set())
for state in ("installed", "enabled", "binding", "bound", "crashed"):
if self._state_field(state) not in printed:
service[self._state_field(state)] = None
service["installed_service_count"] = installed_counts.get(service_user)
named[service_user] = named.get(service_user, 0) + 1
self.results.extend(services.values())
# A stated count the named services do not add up to would leave no
# trace of the rest: the module would log "a total of 0" about a dump
# that said five, or "a total of 1" about one that said two.
for count_user, count in installed_counts.items():
unnamed = count - named.get(count_user, 0)
if unnamed > 0:
self.results.append(self._new_unlisted(count_user, count, unnamed))
@staticmethod
def _describe_state(result: dict) -> str:
if result.get("bound"):
return "enabled and bound"
if result.get("enabled"):
return "enabled"
if result.get("enabled") is None:
return "installed, enabled state not stated"
return "installed, not enabled"
@staticmethod
def _state_field(state: str) -> str:
return {"binding": "binding", "bound": "bound"}.get(state, state)
@staticmethod
def _new_unlisted(user_id: int | None, count: int, unnamed: int) -> dict:
"""The dump's own count for a user, and how many of those services it
did not name.
Every other field stays unknown: the dump named no service to carry it.
"""
record: dict[str, Any] = dict.fromkeys(DumpsysAccessibilityArtifact._FIELDS)
record["user_id"] = user_id
record["installed_service_count"] = count
record["unnamed_service_count"] = unnamed
return record
@staticmethod
def _new_service(component: str, user_id: int | None) -> dict:
record: dict[str, Any] = dict.fromkeys(
DumpsysAccessibilityArtifact._FIELDS, False
)
record["user_id"] = user_id
record["component"] = component
record["package_name"], record["service_name"] = component.split("/", 1)
# Filled in after parsing: the count is per user.
record["installed_service_count"] = None
record["unnamed_service_count"] = None
return record
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,
}
+2 -13
View File
@@ -136,10 +136,6 @@ class Mounts(AndroidArtifact):
mount_options = fields[5].split(",")
super_options = fields[separator + 3].split(",")
options = list(dict.fromkeys(mount_options + super_options))
# fields[5] are the per-mount (VFS) flags, the field after the
# separator the superblock's; a mount is writable only if both
# allow it. The merged list made every read-only overlay a HIGH.
is_read_write = "rw" in mount_options and "rw" in super_options
mount_point = fields[4].replace("\\040", " ")
device = fields[separator + 2].replace("\\040", " ")
filesystem_type = fields[separator + 1]
@@ -160,7 +156,7 @@ class Mounts(AndroidArtifact):
"options_list": options,
"optional_fields": fields[6:separator],
"is_system_partition": is_system,
"is_read_write": is_read_write,
"is_read_write": "rw" in options,
"process_ids": [process_id],
}
)
@@ -195,14 +191,7 @@ class Mounts(AndroidArtifact):
)
# Check for other suspicious mount options
# `rw` sitting in the merged option list is not evidence of
# writability (see is_read_write in parse_mountinfo); every other
# suspicious option is meaningful in either list.
suspicious_opts = [
opt
for opt in options
if opt in SUSPICIOUS_OPTIONS and (opt != "rw" or mount["is_read_write"])
]
suspicious_opts = [opt for opt in options if opt in SUSPICIOUS_OPTIONS]
if suspicious_opts and mount["is_system_partition"]:
if (
"noatime" in mount["mount_options"]
+8 -17
View File
@@ -150,16 +150,6 @@ def check_adb(ctx):
default=[],
help=HELP_MSG_LOAD_MODULE,
)
@click.option(
"--timezone",
"-t",
default=None,
help=(
"IANA timezone name for the device, for example 'Europe/Paris'. "
"Bugreport file timestamps are the device's wall clock; by default the "
"zone is read from persist.sys.timezone in the bugreport itself."
),
)
@click.option("--verbose", "-v", is_flag=True, help=HELP_MSG_VERBOSE_COMMAND)
@click.argument("BUGREPORT_PATH", type=click.Path(exists=True))
@click.pass_context
@@ -170,7 +160,6 @@ def check_bugreport(
list_modules,
module,
load_module,
timezone,
verbose,
bugreport_path,
):
@@ -178,18 +167,12 @@ def check_bugreport(
set_verbose_logging(verbose or _get_verbose(ctx))
custom_modules = _load_custom_modules(load_module)
module_options = {}
if timezone:
module_options["device_timezone"] = timezone
# Always generate hashes as bug reports are small.
cmd = CmdAndroidCheckBugreport(
target_path=bugreport_path,
results_path=output,
ioc_files=iocs,
module_name=module,
module_options=module_options if module_options else None,
hashes=True,
disable_version_check=_get_disable_flags(ctx)[0],
disable_indicator_check=_get_disable_flags(ctx)[1],
@@ -200,6 +183,8 @@ def check_bugreport(
cmd.list_modules()
return
log.info("Checking Android bug report at path: %s", bugreport_path)
try:
cmd.run()
except BadZipFile as exc:
@@ -274,6 +259,8 @@ def check_backup(
cmd.list_modules()
return
log.info("Checking Android backup at path: %s", backup_path)
cmd.run()
cmd.show_alerts_brief()
cmd.show_support_message()
@@ -355,6 +342,8 @@ def check_androidqf(
cmd.list_modules()
return
log.info("Checking AndroidQF acquisition at path: %s", androidqf_path)
cmd.run()
cmd.show_alerts_brief()
cmd.show_disable_adb_warning()
@@ -435,6 +424,8 @@ def check_intrusion_logs(
cmd.list_modules()
return
log.info("Checking intrusion logs at path: %s", logs_path)
cmd.run()
cmd.show_alerts_brief()
cmd.show_support_message()
-2
View File
@@ -82,8 +82,6 @@ class CmdAndroidCheckAndroidQF(Command):
if not self.target_path:
raise NoAndroidQFTargetPath
self.log.info("Checking AndroidQF acquisition at path: %s", self.target_path)
if os.path.isdir(self.target_path):
self.__format = "dir"
parent_path = Path(self.target_path).absolute().parent.as_posix()
-1
View File
@@ -129,7 +129,6 @@ class CmdAndroidCheckBackup(Command):
assert self.target_path is not None # type: ignore[has-type]
# Use a different local variable name to avoid any scoping issues
backup_path: str = self.target_path # type: ignore[has-type]
self.log.info("Checking Android backup at path: %s", backup_path)
if os.path.isfile(backup_path):
self.__type = "ab"
+7 -99
View File
@@ -3,19 +3,13 @@
# Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/
import fnmatch
import io
import logging
import os
from pathlib import Path
from typing import List, Optional
from zipfile import BadZipFile, ZipFile
from zipfile import ZipFile
from mvt.android.artifacts.getprop import GetProp
from mvt.android.modules.bugreport.base import (
DUMPSTATE_ENTRY_POINTS,
BugReportModule,
)
from mvt.android.modules.bugreport.base import BugReportModule
from mvt.common.command import Command
from mvt.common.indicators import Indicators
from mvt.common.module import MVTModule
@@ -93,100 +87,14 @@ class CmdAndroidCheckBugreport(Command):
for file_name in self.__zip.namelist():
self.__files.append(file_name)
if not self._has_dumpstate(self.__files):
nested = self._nested_bugreport(bugreport_zip)
if nested:
self.__zip = nested
self.__files = list(nested.namelist())
@staticmethod
def _has_dumpstate(file_names: List[str]) -> bool:
"""Whether these members carry any of the entry points the bug report
modules read (see `BugReportModule._get_dumpstate_file`)."""
return any(
fnmatch.filter(file_names, pattern) for pattern in DUMPSTATE_ENTRY_POINTS
)
def _nested_bugreport(self, outer: ZipFile) -> Optional[ZipFile]:
"""Descend one level into an OEM wrapper archive.
MIUI / HyperOS hands out a zip of app logs, ANR traces and tcpdump
captures with the real `bugreport-<device>-<timestamp>.zip` nested
inside. Without this descent the outer archive has no entry point,
every module reports it found no files, and the command still exits 0
with an empty result.
"""
candidates = [
name for name in outer.namelist() if name.lower().endswith(".zip")
]
candidates.sort(
key=lambda name: (
"bugreport" not in name.lower() and "dumpstate" not in name.lower()
)
)
for name in candidates:
try:
inner = ZipFile(io.BytesIO(outer.read(name)))
except (BadZipFile, OSError):
continue
if self._has_dumpstate(inner.namelist()):
log.info("Found the bug report nested inside the archive: %s", name)
return inner
inner.close()
return None
def init(self) -> None:
if self.target_path:
self.log.info("Checking Android bug report at path: %s", self.target_path)
if os.path.isfile(self.target_path):
self.from_zip(ZipFile(self.target_path))
elif os.path.isdir(self.target_path):
self.from_dir(self.target_path)
self.log.warning(
"Analysing an unpacked bugreport: file timestamps come from "
"the extraction, not from the device. Analyse the original "
"zip to keep the device's file timestamps."
)
if self.__format:
self._resolve_device_timezone()
def _resolve_device_timezone(self) -> None:
"""Name the device's timezone in module_options unless it is known already.
A bugreport's SYSTEM PROPERTIES section carries persist.sys.timezone.
Zip entry times are the device's wall clock, and modules read them in
this zone; --timezone or check-androidqf's own reading takes precedence.
"""
if self.module_options.get("device_timezone"):
self.log.info("Device timezone: %s", self.module_options["device_timezone"])
if not self.target_path:
return
probe = BugReportModule(log=self.log)
self.module_init(probe)
timezone = None
try:
dumpstate = probe._get_dumpstate_file()
except Exception as exc:
self.log.warning("Could not read the bugreport's dumpstate: %s", exc)
dumpstate = None
if dumpstate:
properties = GetProp()
properties.parse(
BugReportModule.extract_command_section(
dumpstate.decode("utf-8", errors="replace"),
"------ SYSTEM PROPERTIES",
)
)
timezone = properties.get_device_timezone()
if timezone:
self.log.info("Device timezone identified from the bugreport: %s", timezone)
self.module_options["device_timezone"] = timezone
else:
self.log.warning(
"persist.sys.timezone not found in the bugreport; file timestamps "
"are the device's wall clock without a timezone. Pass --timezone "
"to name it."
)
if os.path.isfile(self.target_path):
self.from_zip(ZipFile(self.target_path))
elif os.path.isdir(self.target_path):
self.from_dir(self.target_path)
def module_init(self, module: BugReportModule) -> None: # type: ignore[override]
if self.__format == "zip":
@@ -63,8 +63,6 @@ class CmdAndroidCheckIntrusionLogs(Command):
if not self.target_path:
raise ValueError("No target path specified")
self.log.info("Checking intrusion logs at path: %s", self.target_path)
if not os.path.isdir(self.target_path) and not (
os.path.isfile(self.target_path)
and self.target_path.lower().endswith(".zip")
+9 -47
View File
@@ -6,23 +6,12 @@ import datetime
import fnmatch
import logging
import os
import re
from pathlib import Path
from typing import List, Optional
from zipfile import ZipFile
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from mvt.common.module import ModuleResults, MVTModule
# `------ 0.101s was the duration of 'SOME SECTION' ------`, printed when that
# section finishes and not necessarily between two sections.
SECTION_DURATION = re.compile(r"^-{3,}\s*[0-9.]+s was the duration of", re.IGNORECASE)
# The members a bug report archive can be entered through, in the order
# _get_dumpstate_file() tries them. An archive carrying none of them is not a
# bug report at this level (see CmdAndroidCheckBugreport._has_dumpstate).
DUMPSTATE_ENTRY_POINTS = ("main_entry.txt", "dumpState_*.log", "*/dumpsys.txt")
class BugReportModule(MVTModule):
"""This class provides a base for all Android Bug Report modules."""
@@ -97,9 +86,7 @@ class BugReportModule(MVTModule):
return data
def _get_dumpstate_file(self) -> Optional[bytes]:
main_entry, dumpstate_log, dumpsys_txt = DUMPSTATE_ENTRY_POINTS
main = self._get_files_by_pattern(main_entry)
main = self._get_files_by_pattern("main_entry.txt")
if main:
main_content = self._get_file_content(main[0])
try:
@@ -107,11 +94,11 @@ class BugReportModule(MVTModule):
except KeyError:
return None
dumpstate_logs = self._get_files_by_pattern(dumpstate_log)
dumpstate_logs = self._get_files_by_pattern("dumpState_*.log")
if dumpstate_logs:
return self._get_file_content(dumpstate_logs[0])
dumpsys_files = self._get_files_by_pattern(dumpsys_txt)
dumpsys_files = self._get_files_by_pattern("*/dumpsys.txt")
if dumpsys_files:
return self._get_file_content(dumpsys_files[0])
@@ -134,41 +121,16 @@ class BugReportModule(MVTModule):
in_section = True
continue
if stripped.startswith("------"):
# dumpstate prints a section's timing line when that section
# finishes, which can land in the middle of the one being
# written. Treating it as a boundary truncates the section at
# an arbitrary point, silently.
if SECTION_DURATION.match(stripped):
continue
break
lines.append(line)
return "\n".join(lines)
def _device_timezone(self) -> Optional[datetime.tzinfo]:
"""The device's timezone named in module_options, or None when unknown."""
name = self.module_options.get("device_timezone")
if not name:
return None
try:
return ZoneInfo(name)
except ZoneInfoNotFoundError:
self.log.warning("Unknown device timezone %s", name)
return None
def _get_file_modification_time(self, file_path: str) -> datetime.datetime:
"""When the file was last modified.
A zip entry carries the device's wall clock, so it is returned in the
device's timezone when the bugreport names one and naive otherwise.
An unpacked bugreport's mtime is whatever the extraction left, an
instant returned in UTC.
"""
if self.zip_archive:
file_timetuple = self.zip_archive.getinfo(file_path).date_time
return datetime.datetime(*file_timetuple, tzinfo=self._device_timezone())
if not self.extract_path:
raise ValueError("extract_path is not set")
file_stat = os.stat(os.path.join(self.extract_path, file_path))
return datetime.datetime.fromtimestamp(
file_stat.st_mtime, tz=datetime.timezone.utc
)
return datetime.datetime(*file_timetuple)
else:
if not self.extract_path:
raise ValueError("extract_path is not set")
file_stat = os.stat(os.path.join(self.extract_path, file_path))
return datetime.datetime.fromtimestamp(file_stat.st_mtime)
@@ -48,22 +48,9 @@ class DumpsysAccessibility(DumpsysAccessibilityArtifact, BugReportModule):
)
self.parse(content)
listed = unnamed = 0
for result in self.results:
if result.get("component"):
listed += 1
self.log.info(
'Found accessibility service "%s"', result.get("component")
)
continue
# The operator gets this per user as a LOW alert from
# check_indicators(); here it only has to survive into the summary,
# so that a stated count never reads as "a total of 0".
unnamed += result.get("unnamed_service_count") or 0
self.log.info('Found accessibility service "%s"', result.get("component"))
self.log.info(
"Identified a total of %d accessibility services (%d more stated by "
"the dump without a component name)",
listed,
unnamed,
"Identified a total of %d accessibility services", len(self.results)
)
@@ -4,12 +4,15 @@
# https://license.mvt.re/1.1/
import logging
import datetime
from typing import Optional
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from mvt.common.utils import convert_datetime_to_iso
from .base import BugReportModule
from mvt.common.module_types import ModuleResults
from mvt.android.artifacts.file_timestamps import FileTimestampsArtifact
from mvt.android.artifacts.getprop import GetProp
class BugReportTimestamps(FileTimestampsArtifact, BugReportModule):
@@ -38,28 +41,41 @@ class BugReportTimestamps(FileTimestampsArtifact, BugReportModule):
def run(self) -> None:
filesystem_files = self._get_files_by_pattern("FS/*")
timezone_name = None
dumpstate = self._get_dumpstate_file()
if dumpstate:
section = self.extract_command_section(
dumpstate.decode("utf-8", errors="replace"),
"------ SYSTEM PROPERTIES",
)
properties = GetProp()
properties.parse(section)
timezone_name = properties.get_device_timezone()
timezone = None
if timezone_name:
try:
timezone = ZoneInfo(timezone_name)
except ZoneInfoNotFoundError:
self.log.warning("Unknown device timezone %s", timezone_name)
self.results = []
for file in filesystem_files:
# A zip entry keeps the device's wall clock, read in the device's
# timezone when the command found one (see CmdAndroidCheckBugreport);
# an unpacked bugreport's mtime is read as a UTC instant.
# Only the modification time is available in the zip file metadata.
# The timezone is the local timezone of the machine the phone.
modification_time = self._get_file_modification_time(file)
utc_time = None
if timezone is not None:
utc_time = convert_datetime_to_iso(
modification_time.replace(tzinfo=timezone).astimezone(
datetime.timezone.utc
)
)
self.results.append(
{
"path": file,
"modified_time": convert_datetime_to_iso(
modification_time.replace(tzinfo=None)
),
"modified_time_utc": (
convert_datetime_to_iso(modification_time)
if modification_time.tzinfo
else None
),
"timezone": (
self.module_options.get("device_timezone")
if self.zip_archive
else "UTC"
),
"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"
),
+2 -6
View File
@@ -56,10 +56,6 @@ class CmdCheckIOCS(Command):
if entry not in all_modules:
all_modules.append(entry)
# Read the indicators once, so that a missing indicators file is
# reported even when no stored result matches a module.
iocs = self.iocs
log.info("Checking stored results against provided indicators...")
total_detections = 0
@@ -87,8 +83,8 @@ class CmdCheckIOCS(Command):
log.warning("No result from this module, skipping it")
continue
if iocs.total_ioc_count > 0:
m.indicators = iocs
if self.iocs.total_ioc_count > 0:
m.indicators = self.iocs
m.indicators.log = m.log
try:
+10 -24
View File
@@ -8,7 +8,6 @@ import logging
import os
import sys
from datetime import datetime
from functools import cached_property
from heapq import heappop, heappush
from typing import Any, Optional
@@ -85,18 +84,18 @@ class Command:
self.timeline: ModuleTimeline = []
self.url_results: list[URLResult] = []
# Load IOCs
self._create_storage()
self._setup_logging()
if iocs is not None:
self.iocs = iocs
else:
self.iocs = Indicators(self.log)
self.iocs.load_indicators_files(self.ioc_files)
self.alertstore = AlertStore()
@cached_property
def iocs(self) -> Indicators:
"""Load indicators on first use. Nested commands share their parent's."""
iocs = Indicators(self.log)
iocs.load_indicators_files(self.ioc_files)
return iocs
def _create_storage(self) -> None:
if self.results_path and not os.path.exists(self.results_path):
try:
@@ -711,30 +710,17 @@ class Command:
return ordered
def run(self) -> None:
# The output folder and its command.log exist for a run, so that
# listing modules or rejecting a target leaves nothing behind.
# Resolving the module list can warn, so the log comes first.
self._create_storage()
self._setup_logging()
ordered_modules = self._ordered_modules()
if ordered_modules is None:
return
# Read the indicators once the run is certain to happen, before
# init() does any work on the target, so that a missing indicators
# file is reported first and every module gets the same object.
iocs = self.iocs
self._log_loaded_modules(ordered_modules)
# Commands announce their target from init(), so it goes before the
# module list.
try:
self.init()
except NotImplementedError:
pass
self._log_loaded_modules(ordered_modules)
executed_by_type: dict[type[MVTModule], MVTModule] = {}
for module in ordered_modules:
@@ -754,8 +740,8 @@ class Command:
for dependency, resolved in self._module_dependencies(module)
}
if iocs.total_ioc_count:
m.indicators = iocs
if self.iocs.total_ioc_count:
m.indicators = self.iocs
m.indicators.log = m.log
if self.serial:
+1 -3
View File
@@ -12,9 +12,7 @@ from pydantic_settings import (
YamlConfigSettingsSource,
)
# MVT_CONFIG_FOLDER in the environment relocates the settings file, so that
# a test run or a scripted install never touches the user's own.
MVT_CONFIG_FOLDER = os.environ.get("MVT_CONFIG_FOLDER") or user_config_dir("mvt")
MVT_CONFIG_FOLDER = user_config_dir("mvt")
MVT_CONFIG_PATH = os.path.join(MVT_CONFIG_FOLDER, "config.yaml")
+5 -7
View File
@@ -18,9 +18,7 @@ from appdirs import user_data_dir
from .config import settings
from .url import URL
# MVT_DATA_FOLDER in the environment relocates the downloaded indicators and
# the update-check state kept next to them.
MVT_DATA_FOLDER = os.environ.get("MVT_DATA_FOLDER") or user_data_dir("mvt")
MVT_DATA_FOLDER = user_data_dir("mvt")
MVT_INDICATORS_FOLDER = os.path.join(MVT_DATA_FOLDER, "indicators")
logger = logging.getLogger(__name__)
@@ -73,9 +71,7 @@ class Indicators:
if os.path.isfile(path) and path.lower().endswith(".stix2"):
self.parse_stix2(path)
elif os.path.isdir(path):
for file in glob.glob(
os.path.join(path, "**", "*.stix2"), recursive=True
):
for file in glob.glob(os.path.join(path, "**", "*.stix2"), recursive=True):
self.parse_stix2(file)
else:
self.log.error(
@@ -522,7 +518,9 @@ class Indicators:
the original URL order.
"""
batches = [list(urls) if urls else [] for urls in url_batches]
unique_urls = list(dict.fromkeys(url for urls in batches for url in urls))
unique_urls = list(
dict.fromkeys(url for urls in batches for url in urls)
)
if not unique_urls:
return [None] * len(batches)
+1 -3
View File
@@ -349,9 +349,7 @@ class URL:
return tld_obj
if tld_obj is None:
return ""
# removeprefix, not lstrip: lstrip takes a set of characters, so it ate the
# leading "w"s and dots of any domain ("web.evil.com" -> "eb.evil.com").
return tld_obj.parsed_url.netloc.lower().removeprefix("www.")
return tld_obj.parsed_url.netloc.lower().lstrip("www.")
def get_top_level(self) -> str:
"""Get only the top-level domain from a URL.
+24
View File
@@ -3,6 +3,7 @@
# Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/
import base64
import cProfile
import datetime
import hashlib
@@ -184,6 +185,29 @@ def keys_bytes_to_string(obj: Any) -> Any:
return new_obj
def sanitize_json_data(obj: Any) -> Any:
"""Recursively convert values to JSON-compatible representations.
SQLite and plist records can contain binary and date values nested inside
dictionaries. Preserve binary data as base64 rather than discarding it or
relying on a lossy text decoding during JSON serialization.
"""
if isinstance(obj, datetime.datetime):
return convert_datetime_to_iso(obj)
if isinstance(obj, datetime.date):
return obj.isoformat()
if isinstance(obj, bytes):
return base64.b64encode(obj).decode("ascii")
if isinstance(obj, dict):
return {
sanitize_json_data(key): sanitize_json_data(value)
for key, value in obj.items()
}
if isinstance(obj, (tuple, list, set)):
return [sanitize_json_data(value) for value in obj]
return obj
def get_sha256_from_file_path(file_path: str) -> str:
"""Calculate the SHA256 hash of a file from a file path.
+10 -8
View File
@@ -318,6 +318,8 @@ def check_backup(
if not cmd.resolve_backup_path():
ctx.exit(1)
log.info("Checking iTunes backup located at: %s", cmd.target_path)
cmd.run()
cmd.show_alerts_brief()
cmd.show_support_message()
@@ -384,6 +386,8 @@ def check_fs(
cmd.list_modules()
return
log.info("Checking iOS filesystem located at: %s", dump_path)
cmd.run()
cmd.show_alerts_brief()
cmd.show_support_message()
@@ -445,20 +449,18 @@ def check_sysdiagnose(
custom_modules=custom_modules,
)
# MVT's own module only records the device details; the checks come from
# custom modules, so a run without any must not look like a clean analysis.
if all(module in cmd.modules for module in cmd._available_modules()):
log.warning(
"No forensic sysdiagnose modules have been loaded: MVT's own "
"SysdiagnoseInfo module only records the device details. Install a "
"module package or load a module that declares supported_commands = "
'(("ios", "check-sysdiagnose"),) to check the sysdiagnose.'
if not cmd._available_modules():
raise click.ClickException(
"No custom modules support mvt-ios check-sysdiagnose. "
"Load a module that declares supported_commands = "
"((\"ios\", \"check-sysdiagnose\"),)."
)
if list_modules:
cmd.list_modules()
return
log.info("Checking iOS sysdiagnose at path: %s", sysdiagnose_path)
cmd.run()
cmd.show_alerts_brief()
cmd.show_support_message()
-3
View File
@@ -59,9 +59,6 @@ class CmdIOSCheckBackup(Command):
self.name = "check-backup"
self.modules = BACKUP_MODULES + MIXED_MODULES
def init(self) -> None:
self.log.info("Checking iTunes backup located at: %s", self.target_path)
def resolve_backup_path(self) -> bool:
target_path = getattr(self, "target_path", None)
if not isinstance(target_path, str) or not target_path:
-3
View File
@@ -51,8 +51,5 @@ class CmdIOSCheckFS(Command):
self.name = "check-fs"
self.modules = FS_MODULES + MIXED_MODULES
def init(self) -> None:
self.log.info("Checking iOS filesystem located at: %s", self.target_path)
def module_init(self, module):
module.is_fs_dump = True
+2 -27
View File
@@ -7,9 +7,7 @@ import json
import logging
import os
import shutil
import sys
import tarfile
import zlib
from pathlib import Path, PurePosixPath
from tempfile import TemporaryDirectory
from typing import Any, Optional
@@ -18,8 +16,6 @@ from mvt.common.command import Command
from mvt.common.indicators import Indicators
from mvt.common.module import MVTModule
from .modules.sysdiagnose import SYSDIAGNOSE_MODULES
log = logging.getLogger(__name__)
@@ -56,7 +52,6 @@ class CmdIOSCheckSysdiagnose(Command):
)
self.platform = "ios"
self.name = "check-sysdiagnose"
self.modules = SYSDIAGNOSE_MODULES
self.sysdiagnose_format: Optional[str] = None
self.sysdiagnose_archive: Optional[tarfile.TarFile] = None
self.sysdiagnose_files: list[str] = []
@@ -81,15 +76,11 @@ class CmdIOSCheckSysdiagnose(Command):
if not self.target_path:
raise ValueError("A sysdiagnose path is required")
self.log.info("Checking iOS sysdiagnose at path: %s", self.target_path)
if os.path.isdir(self.target_path):
self.sysdiagnose_format = "dir"
parent_path = Path(self.target_path).absolute().parent
for root, _, filenames in os.walk(self.target_path):
for filename in filenames:
if filename.startswith("._"):
continue
absolute_path = os.path.join(root, filename)
file_path = os.path.relpath(absolute_path, parent_path)
self.sysdiagnose_files.append(file_path)
@@ -103,19 +94,8 @@ class CmdIOSCheckSysdiagnose(Command):
self.log.info("Parsing sysdiagnose archive. This might take a while...")
self.sysdiagnose_format = "tar"
try:
self.sysdiagnose_archive = tarfile.open(self.target_path, "r:gz")
self._extract_sysdiagnose_archive()
except (tarfile.ReadError, EOFError, zlib.error, OSError) as exc:
# A truncated archive ends in EOFError from gzip, which Click would
# otherwise report as a bare "Aborted!" with no reason.
self.log.critical(
"Unable to read the sysdiagnose archive %s: %s. "
"The file may be truncated or not a gzip-compressed tarball.",
self.target_path,
exc,
)
sys.exit(1)
self.sysdiagnose_archive = tarfile.open(self.target_path, "r:gz")
self._extract_sysdiagnose_archive()
def _extract_sysdiagnose_archive(self) -> None:
archive = self.sysdiagnose_archive
@@ -139,11 +119,6 @@ class CmdIOSCheckSysdiagnose(Command):
if not member_path.parts:
continue
# AppleDouble sidecars (._name) carry a file's extended attributes,
# not sysdiagnose content. Device archives hold hundreds of them;
# bsdtar hides them from listings, tarfile does not.
if member_path.name.startswith("._"):
continue
archive_roots.add(member_path.parts[0])
if member.isdir():
+1 -2
View File
@@ -16,8 +16,7 @@ from mvt.common.module import MVTModule
from .modules.backup import BACKUP_MODULES
from .modules.fs import FS_MODULES
from .modules.mixed import MIXED_MODULES
from .modules.sysdiagnose import SYSDIAGNOSE_MODULES
IOS_CHECK_IOCS_MODULES: list[type[MVTModule]] = (
BACKUP_MODULES + FS_MODULES + MIXED_MODULES + SYSDIAGNOSE_MODULES
BACKUP_MODULES + FS_MODULES + MIXED_MODULES
)
-8
View File
@@ -1271,13 +1271,5 @@
{
"version": "26.6.1",
"build": "23G83"
},
{
"version": "26.6.2",
"build": "23G90"
},
{
"version": "27.0",
"build": "24A437"
}
]
+1 -1
View File
@@ -52,7 +52,7 @@ class MVTEncryptedBackup(EncryptedBackup):
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.10.0.
internals). Pinned to iphone_backup_decrypt==0.9.0.
"""
def __init__(self, *, backup_directory, passphrase=None, derived_key=None):
+58 -49
View File
@@ -11,7 +11,7 @@ from mvt.common.module_types import (
ModuleResults,
ModuleSerializedResult,
)
from mvt.common.utils import convert_mactime_to_iso
from mvt.common.utils import convert_mactime_to_iso, sanitize_json_data
from ..base import IOSExtraction
@@ -90,54 +90,63 @@ class Calendar(IOSExtraction):
"""
conn = self._open_sqlite_db(self.file_path)
cur = conn.cursor()
cur.execute(
"""
SELECT
CalendarItem.ROWID as "id",
CalendarItem.summary as "summary",
CalendarItem.description as "description",
CalendarItem.start_date as "start_date",
CalendarItem.end_date as "end_date",
CalendarItem.all_day as "all_day",
CalendarItem.calendar_id as "calendar_id",
CalendarItem.organizer_id as "organizer_id",
CalendarItem.url as "url",
CalendarItem.last_modified as "last_modified",
CalendarItem.external_id as "external_id",
CalendarItem.external_mod_tag as "external_mod_tag",
CalendarItem.unique_identifier as "unique_identifier",
CalendarItem.hidden as "hidden",
CalendarItem.UUID as "uuid",
CalendarItem.creation_date as "creation_date",
CalendarItem.action as "action",
CalendarItem.created_by_id as "created_by_id",
Participant.UUID as "participant_uuid",
Participant.email as "participant_email",
Participant.phone_number as "participant_phone",
Participant.comment as "participant_comment",
Participant.last_modified as "participant_last_modified"
FROM CalendarItem
LEFT JOIN Participant ON Participant.ROWID = CalendarItem.organizer_id;
"""
)
names = [description[0] for description in cur.description]
for item in cur:
entry = {}
for index, value in enumerate(item):
if names[index] in self.timestamps:
if value is None or isinstance(value, str):
entry[names[index]] = value
else:
entry[names[index]] = convert_mactime_to_iso(value)
else:
entry[names[index]] = value
self.results.append(entry)
cur.close()
conn.close()
try:
cur.execute("PRAGMA table_info(CalendarItem);")
calendar_columns = [row[1] for row in cur]
cur.execute("PRAGMA table_info(Participant);")
participant_columns = [row[1] for row in cur]
selected = [
f'c."{column}" AS "calendar_{column}"' for column in calendar_columns
] + [
f'p."{column}" AS "participant_{column}"'
for column in participant_columns
]
cur.execute(
f"SELECT {', '.join(selected)} FROM CalendarItem c "
"LEFT JOIN Participant p ON p.ROWID = c.organizer_id;"
)
names = [description[0] for description in cur.description]
for row in cur:
raw = sanitize_json_data(dict(zip(names, row)))
entry = {
"id": raw.get("calendar_ROWID"),
"summary": raw.get("calendar_summary"),
"description": raw.get("calendar_description"),
"all_day": raw.get("calendar_all_day"),
"calendar_id": raw.get("calendar_calendar_id"),
"organizer_id": raw.get("calendar_organizer_id"),
"url": raw.get("calendar_url"),
"external_id": raw.get("calendar_external_id"),
"external_mod_tag": raw.get("calendar_external_mod_tag"),
"unique_identifier": raw.get("calendar_unique_identifier"),
"hidden": raw.get("calendar_hidden"),
"uuid": raw.get("calendar_UUID"),
"action": raw.get("calendar_action"),
"created_by_id": raw.get("calendar_created_by_id"),
"participant_uuid": raw.get("participant_UUID"),
"participant_email": raw.get("participant_email"),
"participant_phone": raw.get("participant_phone_number"),
"participant_comment": raw.get("participant_comment"),
"record": raw,
}
timestamp_sources = {
"start_date": "calendar_start_date",
"end_date": "calendar_end_date",
"last_modified": "calendar_last_modified",
"creation_date": "calendar_creation_date",
"participant_last_modified": "participant_last_modified",
}
for output, source in timestamp_sources.items():
value = raw.get(source)
entry[output] = (
convert_mactime_to_iso(value)
if value is not None and not isinstance(value, str)
else value
)
self.results.append(entry)
finally:
cur.close()
conn.close()
def run(self) -> None:
self._find_ios_database(
+26 -27
View File
@@ -7,7 +7,7 @@ import logging
from typing import Optional
from mvt.common.module_types import ModuleAtomicResult, ModuleSerializedResult
from mvt.common.utils import convert_mactime_to_iso
from mvt.common.utils import convert_mactime_to_iso, sanitize_json_data
from ..base import IOSExtraction
@@ -57,31 +57,30 @@ class Calls(IOSExtraction):
return
conn = self._open_sqlite_db(self.file_path)
cur = conn.cursor()
cur.execute(
"""
SELECT
ZDATE, ZDURATION, ZLOCATION, ZADDRESS, ZSERVICE_PROVIDER
FROM ZCALLRECORD;
"""
)
# names = [description[0] for description in cur.description]
for row in cur:
self.results.append(
{
"isodate": convert_mactime_to_iso(row[0]),
"duration": row[1],
"location": row[2],
"number": (
row[3].decode("utf-8")
if isinstance(row[3], bytes)
else row[3]
),
"provider": row[4],
}
)
cur.close()
conn.close()
try:
cur.execute("SELECT * FROM ZCALLRECORD;")
names = [description[0] for description in cur.description]
for row in cur:
original = dict(zip(names, row))
address = original.get("ZADDRESS")
if isinstance(address, bytes):
address = address.decode("utf-8", errors="replace")
raw = sanitize_json_data(original)
date = raw.get("ZDATE")
self.results.append(
{
"isodate": (
convert_mactime_to_iso(date) if date is not None else ""
),
"duration": raw.get("ZDURATION"),
"location": raw.get("ZLOCATION"),
"number": address,
"provider": raw.get("ZSERVICE_PROVIDER"),
"call": raw,
}
)
finally:
cur.close()
conn.close()
self.log.info("Extracted a total of %d calls", len(self.results))
+38 -20
View File
@@ -8,6 +8,7 @@ import sqlite3
from typing import Optional
from mvt.common.module_types import ModuleResults
from mvt.common.utils import sanitize_json_data
from ..base import IOSExtraction
@@ -51,28 +52,45 @@ class Contacts(IOSExtraction):
conn = self._open_sqlite_db(self.file_path)
cur = conn.cursor()
try:
try:
cur.execute(
"""
SELECT
multi.value, person.first, person.middle, person.last,
person.organization
FROM ABPerson person, ABMultiValue multi
WHERE person.rowid = multi.record_id and multi.value not null
ORDER by person.rowid ASC;
"""
)
except sqlite3.OperationalError as e:
self.log.info("Error while reading the contact table: %s", e)
return None
cur.execute("PRAGMA table_info(ABPerson);")
person_columns = [row[1] for row in cur]
cur.execute("PRAGMA table_info(ABMultiValue);")
multivalue_columns = [row[1] for row in cur]
if not person_columns or not multivalue_columns:
return
select_columns = [
f'p."{column}" AS "person_{column}"' for column in person_columns
] + [
f'm."{column}" AS "multivalue_{column}"'
for column in multivalue_columns
]
cur.execute(
f"SELECT {', '.join(select_columns)} "
"FROM ABPerson p LEFT JOIN ABMultiValue m "
"ON p.ROWID = m.record_id ORDER BY p.ROWID, m.ROWID;"
)
names = [description[0] for description in cur.description]
for row in cur:
new_contact = {}
for index, value in enumerate(row):
new_contact[names[index]] = value
self.results.append(new_contact)
raw = sanitize_json_data(dict(zip(names, row)))
self.results.append(
{
"value": raw.get("multivalue_value"),
"first": raw.get("person_First"),
"middle": raw.get("person_Middle"),
"last": raw.get("person_Last"),
"organization": raw.get("person_Organization"),
# Keep the capitalization returned by the previous
# query for consumers of existing JSON output.
"First": raw.get("person_First"),
"Middle": raw.get("person_Middle"),
"Last": raw.get("person_Last"),
"Organization": raw.get("person_Organization"),
"contact": raw,
}
)
except sqlite3.OperationalError as exc:
self.log.info("Error while reading the contact table: %s", exc)
finally:
cur.close()
conn.close()
+19 -1
View File
@@ -14,7 +14,7 @@ from mvt.common.module_types import (
ModuleResults,
ModuleSerializedResult,
)
from mvt.common.utils import convert_mactime_to_iso
from mvt.common.utils import convert_mactime_to_iso, sanitize_json_data
from ..base import IOSExtraction
from .whatsapp_contacts import WhatsappContacts
@@ -547,4 +547,22 @@ class InteractionC(IOSExtraction):
self._postprocess_results()
# Preserve every column from the underlying interaction row. The
# joined compatibility queries above intentionally expose a stable
# result schema but otherwise discard useful forensic fields.
conn = self._open_sqlite_db(self.file_path)
cur = conn.cursor()
try:
cur.execute("SELECT * FROM ZINTERACTIONS;")
names = [description[0] for description in cur.description]
interactions = {}
for row in cur:
raw = sanitize_json_data(dict(zip(names, row)))
interactions[raw.get("Z_PK")] = raw
finally:
cur.close()
conn.close()
for result in self.results:
result["interaction"] = interactions.get(result.get("table_id"), {})
self.log.info("Extracted a total of %d InteractionC events", len(self.results))
@@ -7,6 +7,8 @@ import logging
from typing import Optional
from mvt.common.module_types import ModuleResults
from mvt.common.utils import convert_mactime_to_iso, sanitize_json_data
from ..net_base import NetBase
DATAUSAGE_BACKUP_IDS = [
@@ -42,6 +44,77 @@ class Datausage(NetBase):
results=results,
)
def _extract_net_data(self):
assert self.file_path is not None
conn = self._open_sqlite_db(self.file_path)
cur = conn.cursor()
try:
cur.execute("PRAGMA table_info(ZPROCESS);")
process_columns = [row[1] for row in cur]
cur.execute("PRAGMA table_info(ZLIVEUSAGE);")
live_columns = [row[1] for row in cur]
selected = [
f'p."{column}" AS "process_{column}"' for column in process_columns
] + [f'l."{column}" AS "live_{column}"' for column in live_columns]
cur.execute(
f"SELECT {', '.join(selected)} FROM ZLIVEUSAGE l "
"LEFT JOIN ZPROCESS p ON l.ZHASPROCESS = p.Z_PK;"
)
names = [description[0] for description in cur.description]
rows = [dict(zip(names, row)) for row in cur]
process_only = [
f'p."{column}" AS "process_{column}"' for column in process_columns
] + [f'NULL AS "live_{column}"' for column in live_columns]
cur.execute(
f"SELECT {', '.join(process_only)} FROM ZPROCESS p "
"WHERE NOT EXISTS ("
"SELECT 1 FROM ZLIVEUSAGE l WHERE l.ZHASPROCESS = p.Z_PK);"
)
names = [description[0] for description in cur.description]
rows.extend(dict(zip(names, row)) for row in cur)
finally:
cur.close()
conn.close()
for raw_row in rows:
first_timestamp = raw_row.get("process_ZFIRSTTIMESTAMP")
process_timestamp = raw_row.get("process_ZTIMESTAMP")
live_timestamp = raw_row.get("live_ZTIMESTAMP")
first_isodate = (
convert_mactime_to_iso(first_timestamp)
if first_timestamp
else first_timestamp
)
isodate = (
convert_mactime_to_iso(process_timestamp)
if process_timestamp
else process_timestamp
)
live_isodate = (
convert_mactime_to_iso(live_timestamp)
if live_timestamp
else first_isodate
)
self.results.append(
{
"first_isodate": first_isodate,
"isodate": isodate,
"proc_name": raw_row.get("process_ZPROCNAME"),
"bundle_id": raw_row.get("process_ZBUNDLENAME"),
"proc_id": raw_row.get("process_Z_PK"),
"wifi_in": raw_row.get("live_ZWIFIIN"),
"wifi_out": raw_row.get("live_ZWIFIOUT"),
"wwan_in": raw_row.get("live_ZWWANIN"),
"wwan_out": raw_row.get("live_ZWWANOUT"),
"live_id": raw_row.get("live_Z_PK"),
"live_proc_id": raw_row.get("live_ZHASPROCESS"),
"live_isodate": live_isodate,
"record": sanitize_json_data(raw_row),
}
)
self.log.info("Extracted information on %d processes", len(self.results))
def run(self) -> None:
self._find_ios_database(
backup_ids=DATAUSAGE_BACKUP_IDS, root_paths=DATAUSAGE_ROOT_PATHS
@@ -15,7 +15,11 @@ from mvt.common.module_types import (
ModuleResults,
ModuleSerializedResult,
)
from mvt.common.utils import convert_mactime_to_iso, keys_bytes_to_string
from mvt.common.utils import (
convert_mactime_to_iso,
keys_bytes_to_string,
sanitize_json_data,
)
from ..base import IOSExtraction
@@ -94,85 +98,76 @@ class SafariBrowserState(IOSExtraction):
def _process_browser_state_db(self, db_path):
self._recover_sqlite_db_if_needed(db_path)
conn = self._open_sqlite_db(db_path)
cur = conn.cursor()
try:
try:
cur.execute(
"""
SELECT
tabs.title,
tabs.url,
tabs.user_visible_url,
tabs.last_viewed_time,
tab_sessions.session_data
SELECT tabs.*, tab_sessions.session_data AS session_data
FROM tabs
JOIN tab_sessions ON tabs.uuid = tab_sessions.tab_uuid
LEFT JOIN tab_sessions ON tabs.uuid = tab_sessions.tab_uuid
ORDER BY tabs.last_viewed_time;
"""
"""
)
names = [description[0] for description in cur.description]
except sqlite3.OperationalError:
# Old version iOS <12 likely
# Older databases store session_data directly in tabs.
try:
cur.execute(
"""
SELECT
title, url, user_visible_url, last_viewed_time, session_data
FROM tabs
ORDER BY last_viewed_time;
"""
)
except sqlite3.OperationalError as e:
self.log.error(f"Error executing query: {e}")
cur.execute("SELECT * FROM tabs ORDER BY last_viewed_time;")
names = [description[0] for description in cur.description]
except sqlite3.OperationalError as exc:
self.log.error("Error executing query: %s", exc)
return
for row in cur:
raw = dict(zip(names, row))
session_entries = []
if row[4]:
session_blob = raw.get("session_data")
if session_blob:
# Skip a 4 byte header before the plist content.
session_plist = row[4][4:]
session_plist = session_blob[4:]
session_data = {}
try:
session_data = plistlib.load(io.BytesIO(session_plist))
session_data = keys_bytes_to_string(session_data)
except plistlib.InvalidFileException:
except (plistlib.InvalidFileException, ValueError, TypeError):
pass
if "SessionHistoryEntries" in session_data.get("SessionHistory", {}):
for session_entry in session_data["SessionHistory"].get(
"SessionHistoryEntries"
):
self._session_history_count += 1
data_length = 0
if "SessionHistoryEntryData" in session_entry:
data_length = len(
session_entry.get("SessionHistoryEntryData")
)
session_entries.append(
{
"entry_title": session_entry.get(
"SessionHistoryEntryOriginalURL"
),
"entry_url": session_entry.get(
"SessionHistoryEntryURL"
),
"data_length": data_length,
}
)
for session_entry in session_data.get("SessionHistory", {}).get(
"SessionHistoryEntries", []
):
self._session_history_count += 1
entry_data = session_entry.get("SessionHistoryEntryData")
session_entries.append(
{
"entry_title": session_entry.get(
"SessionHistoryEntryOriginalURL"
),
"entry_url": session_entry.get(
"SessionHistoryEntryURL"
),
"data_length": (
len(entry_data) if entry_data is not None else 0
),
}
)
last_viewed = raw.get("last_viewed_time")
self.results.append(
{
"tab_title": row[0],
"tab_url": row[1],
"tab_visible_url": row[2],
"last_viewed_timestamp": convert_mactime_to_iso(row[3]),
"tab_title": raw.get("title"),
"tab_url": raw.get("url"),
"tab_visible_url": raw.get("user_visible_url"),
"last_viewed_timestamp": (
convert_mactime_to_iso(last_viewed)
if last_viewed is not None
else ""
),
"session_data": session_entries,
"safari_browser_state_db": os.path.relpath(
db_path, self.target_path
),
"tab": sanitize_json_data(raw),
}
)
finally:
+39 -33
View File
@@ -13,7 +13,11 @@ from mvt.common.module_types import (
ModuleSerializedResult,
)
from mvt.common.url import URL
from mvt.common.utils import convert_mactime_to_datetime, convert_mactime_to_iso
from mvt.common.utils import (
convert_mactime_to_datetime,
convert_mactime_to_iso,
sanitize_json_data,
)
from ..base import IOSExtraction
@@ -133,39 +137,41 @@ class SafariHistory(IOSExtraction):
self._recover_sqlite_db_if_needed(history_path)
conn = self._open_sqlite_db(history_path)
cur = conn.cursor()
cur.execute(
"""
SELECT
history_items.id,
history_items.url,
history_visits.id,
history_visits.visit_time,
history_visits.redirect_source,
history_visits.redirect_destination
FROM history_items
JOIN history_visits ON history_visits.history_item = history_items.id
ORDER BY history_visits.visit_time;
"""
)
for row in cur:
self.results.append(
{
"id": row[0],
"url": row[1],
"visit_id": row[2],
"timestamp": row[3],
"isodate": convert_mactime_to_iso(row[3]),
"redirect_source": row[4],
"redirect_destination": row[5],
"safari_history_db": os.path.relpath(
history_path, self.target_path
),
}
try:
cur.execute("PRAGMA table_info(history_items);")
item_columns = [row[1] for row in cur]
cur.execute("PRAGMA table_info(history_visits);")
visit_columns = [row[1] for row in cur]
selected = [
f'i."{column}" AS "item_{column}"' for column in item_columns
] + [f'v."{column}" AS "visit_{column}"' for column in visit_columns]
cur.execute(
f"SELECT {', '.join(selected)} FROM history_items i "
"JOIN history_visits v ON v.history_item = i.id "
"ORDER BY v.visit_time;"
)
cur.close()
conn.close()
names = [description[0] for description in cur.description]
for row in cur:
raw = sanitize_json_data(dict(zip(names, row)))
timestamp = raw.get("visit_visit_time")
self.results.append(
{
"id": raw.get("item_id"),
"url": raw.get("item_url"),
"visit_id": raw.get("visit_id"),
"timestamp": timestamp,
"isodate": convert_mactime_to_iso(timestamp),
"redirect_source": raw.get("visit_redirect_source"),
"redirect_destination": raw.get("visit_redirect_destination"),
"safari_history_db": os.path.relpath(
history_path, self.target_path
),
"record": raw,
}
)
finally:
cur.close()
conn.close()
def run(self) -> None:
if self.is_backup:
+37 -114
View File
@@ -12,7 +12,7 @@ from mvt.common.module_types import (
ModuleResults,
ModuleSerializedResult,
)
from mvt.common.utils import convert_unix_to_iso
from mvt.common.utils import convert_unix_to_iso, sanitize_json_data
from ..base import IOSExtraction
@@ -103,126 +103,49 @@ class TCC(IOSExtraction):
def process_db(self, file_path):
conn = self._open_sqlite_db(file_path)
cur = conn.cursor()
db_version = "v3"
try:
cur.execute(
"""SELECT
service, client, client_type, auth_value,
auth_reason, last_modified
FROM access;"""
)
except sqlite3.OperationalError:
# v2 version
try:
cur.execute(
"""SELECT
service, client, client_type, allowed,
prompt_count, last_modified
FROM access;"""
cur.execute("SELECT * FROM access;")
except sqlite3.OperationalError as exc:
self.log.error("Error parsing TCC database: %s", exc)
return
names = [description[0] for description in cur.description]
for row in cur:
record = sanitize_json_data(dict(zip(names, row)))
client_type = record.get("client_type")
record["client_type_value"] = client_type
record["client_type"] = (
"bundle_id" if client_type == 0 else "absolute_path"
)
db_version = "v2"
except sqlite3.OperationalError:
try:
cur.execute(
"""SELECT
service, client, client_type, allowed,
prompt_count
FROM access;"""
)
db_version = "v1"
except sqlite3.OperationalError as e:
self.log.error(f"Error parsing TCC database: {e}")
for row in cur:
service = row[0]
client = row[1]
client_type = row[2]
client_type_desc = "bundle_id" if client_type == 0 else "absolute_path"
if db_version == "v3":
auth_value = row[3]
auth_value_desc = AUTH_VALUES.get(auth_value, "")
auth_reason = row[4]
auth_reason_desc = AUTH_REASONS.get(auth_reason, "unknown")
last_modified = convert_unix_to_iso(row[5])
if service in ["kTCCServiceMicrophone", "kTCCServiceCamera"]:
device = (
"microphone" if service == "kTCCServiceMicrophone" else "camera"
if "auth_value" in record:
auth_value = record["auth_value"]
record["auth_value_value"] = auth_value
record["auth_value"] = AUTH_VALUES.get(auth_value, "unknown")
auth_reason = record.get("auth_reason")
record["auth_reason_desc"] = AUTH_REASONS.get(
auth_reason, "unknown"
)
self.log.info(
'Found client "%s" with access %s to %s on %s by %s',
client,
auth_value_desc,
device,
last_modified,
auth_reason_desc,
elif "allowed" in record:
allowed = record["allowed"]
record["allowed_value"] = AUTH_VALUE_OLD.get(allowed, "unknown")
if record.get("last_modified") is not None:
record["last_modified_value"] = record["last_modified"]
record["last_modified"] = convert_unix_to_iso(
record["last_modified"]
)
if record.get("last_reminded") is not None:
record["last_reminded_value"] = record["last_reminded"]
record["last_reminded"] = convert_unix_to_iso(
record["last_reminded"]
)
self.results.append(
{
"service": service,
"client": client,
"client_type": client_type_desc,
"auth_value": auth_value_desc,
"auth_reason_desc": auth_reason_desc,
"last_modified": last_modified,
}
)
else:
allowed_value = row[3]
allowed_desc = AUTH_VALUE_OLD.get(allowed_value, "")
prompt_count = row[4]
if db_version == "v2":
last_modified = convert_unix_to_iso(row[5])
if service in ["kTCCServiceMicrophone", "kTCCServiceCamera"]:
device = "camera"
if service == "kTCCServiceMicrophone":
device = "microphone"
self.log.info(
'Found client "%s" with access %s to %s at %s',
client,
allowed_desc,
device,
last_modified,
)
self.results.append(
{
"service": service,
"client": client,
"client_type": client_type_desc,
"allowed_value": allowed_desc,
"prompt_count": prompt_count,
"last_modified": last_modified,
}
)
else:
if service in ["kTCCServiceMicrophone", "kTCCServiceCamera"]:
device = "camera"
if service == "kTCCServiceMicrophone":
device = "microphone"
self.log.info(
'Found client "%s" with access %s to %s',
client,
allowed_desc,
device,
)
self.results.append(
{
"service": service,
"client": client,
"client_type": client_type_desc,
"allowed_value": allowed_desc,
"prompt_count": prompt_count,
}
)
cur.close()
conn.close()
self.results.append(record)
finally:
cur.close()
conn.close()
def run(self) -> None:
self._find_ios_database(backup_ids=TCC_BACKUP_IDS, root_paths=TCC_ROOT_PATHS)
@@ -13,7 +13,7 @@ from mvt.common.module_types import (
ModuleResults,
ModuleSerializedResult,
)
from mvt.common.utils import convert_unix_to_iso
from mvt.common.utils import convert_unix_to_iso, sanitize_json_data
from ..base import IOSExtraction
@@ -85,52 +85,45 @@ class WebkitResourceLoadStatistics(IOSExtraction):
try:
try:
cur.execute("PRAGMA table_info(ObservedDomains);")
available_columns = {row[1] for row in cur}
required_columns = [
"domainID",
"registrableDomain",
"lastSeen",
"hadUserInteraction",
]
if not set(required_columns).issubset(available_columns):
return
optional_columns = [
column
for column in [
"mostRecentUserInteractionTime",
"mostRecentWebPushInteractionTime",
]
if column in available_columns
]
selected_columns = required_columns + optional_columns
cur.execute(
f"SELECT {', '.join(selected_columns)} FROM ObservedDomains;"
)
cur.execute("SELECT * FROM ObservedDomains;")
except sqlite3.OperationalError:
return
names = [description[0] for description in cur.description]
required_columns = {
"domainID",
"registrableDomain",
"lastSeen",
"hadUserInteraction",
}
if not required_columns.issubset(names):
return
for row in cur:
raw = sanitize_json_data(dict(zip(names, row)))
last_seen = raw.get("lastSeen")
result = {
"domain_id": row[0],
"registrable_domain": row[1],
"last_seen": row[2],
"had_user_interaction": bool(row[3]),
"last_seen_isodate": convert_unix_to_iso(row[2]),
"domain_id": raw.get("domainID"),
"registrable_domain": raw.get("registrableDomain"),
"last_seen": last_seen,
"had_user_interaction": bool(raw.get("hadUserInteraction")),
"last_seen_isodate": convert_unix_to_iso(last_seen),
"domain": domain,
"path": path,
"record": raw,
}
for index, column in enumerate(optional_columns, start=4):
field = {
"mostRecentUserInteractionTime": (
"most_recent_user_interaction_time"
),
"mostRecentWebPushInteractionTime": (
"most_recent_web_push_interaction_time"
),
}[column]
timestamp = row[index]
for column, field in (
(
"mostRecentUserInteractionTime",
"most_recent_user_interaction_time",
),
(
"mostRecentWebPushInteractionTime",
"most_recent_web_push_interaction_time",
),
):
if column not in raw:
continue
timestamp = raw[column]
result[field] = timestamp
if timestamp is not None and timestamp >= 0:
result[f"{field}_isodate"] = convert_unix_to_iso(timestamp)
@@ -3,9 +3,4 @@
# Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/
from mvt.common.module import MVTModule
from .base import SysdiagnoseExtraction
from .sysdiagnose_info import SysdiagnoseInfo
SYSDIAGNOSE_MODULES: list[type[MVTModule]] = [SysdiagnoseInfo]
@@ -1,228 +0,0 @@
# 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 json
import logging
import os
import plistlib
import re
import sqlite3
import tempfile
from datetime import datetime
from pathlib import Path
from typing import Optional
from mvt.common.module_types import ModuleResults
from mvt.common.utils import convert_datetime_to_iso
from mvt.ios.versions import (
find_version_by_build,
get_device_desc_from_id,
is_ios_version_outdated,
)
from .base import SysdiagnoseExtraction
# The fields dumpsys prints in the remotectl dump state and the mobile
# activation request which are worth a log line of their own.
LOGGED_FIELDS = (
"ProductName",
"ProductType",
"SerialNumber",
"OSVersion",
"RegionCode",
"IMEI",
"BuildVersion",
)
class SysdiagnoseInfo(SysdiagnoseExtraction):
"""Extract details about the device and the sysdiagnose itself.
The fields come from four files of the archive: the remotectl dump state
(product type, OS version, serial number, region and the rest of its
Properties block), the mobile activation request (UDID, IMEI, MEID and the
OS build), the App Store daemon database (the Apple account name and email)
and sysdiagnose.log (the archive's original file name and creation time).
Newer iOS versions no longer include the App Store daemon database in a
sysdiagnose; it is still read for the analysis of older archives.
"""
def __init__(
self,
file_path: Optional[str] = None,
target_path: Optional[str] = None,
results_path: Optional[str] = None,
module_options: Optional[dict] = None,
log: logging.Logger = logging.getLogger(__name__),
results: Optional[ModuleResults] = None,
) -> None:
super().__init__(
file_path=file_path,
target_path=target_path,
results_path=results_path,
module_options=module_options,
log=log,
results=results,
)
self.results: dict = results if results is not None else {}
def _copy_sqlite_db(self, file_path: str, directory: str) -> str:
"""Copy a database and its WAL sidecars out of the archive.
A database dumped mid-transaction keeps its latest rows in the -wal
file next to it, which SQLite only reads when both sit in the same
directory under the same name.
"""
available_files = self.tar_files if self.tar else self.files
for suffix in ("", "-wal", "-shm"):
if suffix and f"{file_path}{suffix}" not in available_files:
continue
copy_path = os.path.join(directory, f"{Path(file_path).name}{suffix}")
with open(copy_path, "wb") as handle:
handle.write(self._get_file_content(f"{file_path}{suffix}"))
return os.path.join(directory, Path(file_path).name)
def _process_appstored(self, file_path: str) -> None:
self.log.info("Found App Store daemon database at: %s", file_path)
with tempfile.TemporaryDirectory(prefix="mvt_sqlite_") as directory:
db_path = Path(self._copy_sqlite_db(file_path, directory)).resolve()
conn = sqlite3.connect(f"{db_path.as_uri()}?mode=ro", uri=True)
try:
self._read_appstored(conn)
finally:
conn.close()
def _read_appstored(self, conn: sqlite3.Connection) -> None:
cur = conn.cursor()
# The account name sits in an opaque structure of every asset row.
try:
rows = cur.execute("SELECT sinfs_data FROM asset;").fetchall()
except sqlite3.DatabaseError as exc:
self.log.debug("Unable to read the asset table: %s", exc)
rows = []
for (sinfs_data,) in rows:
try:
sinf = plistlib.loads(sinfs_data)[0]["sinf"]
except (plistlib.InvalidFileException, IndexError, KeyError, TypeError):
continue
match = re.search(rb"name(.*?)\x00", sinf)
if match:
self.results["Account Name"] = match.group(1).decode(
"utf-8", errors="replace"
)
break
try:
row = cur.execute(
"SELECT store_account_name FROM job_software "
"WHERE store_account_name IS NOT NULL LIMIT 1;"
).fetchone()
except sqlite3.DatabaseError as exc:
self.log.debug("Unable to read the job_software table: %s", exc)
return
if row:
self.results["Email Address"] = row[0]
def _process_activation_log(self, file_path: str) -> None:
self.log.info("Found mobile activation request at: %s", file_path)
content = self._get_file_content(file_path)
match = re.search(rb"BODY:\s+({.+?})\s", content, re.MULTILINE)
if not match:
return
try:
body = json.loads(match.group(1))
except json.JSONDecodeError as exc:
self.log.warning("Unable to parse the activation request body: %s", exc)
return
self.results.update(
{
"SerialNumber": body.get("serial-number"),
"ProductType": body.get("productType"),
"ProductName": body.get("productName"),
"IMEI": body.get("imei"),
"ProductVersion": body.get("os-version"),
"UniqueIdentifier": body.get("udid"),
"MEID": body.get("meid"),
"BuildVersion": body.get("os-build"),
}
)
def _process_dumpstate(self, file_path: str) -> None:
self.log.info("Found remotectl dump state at: %s", file_path)
content = self._get_file_content(file_path).decode("utf-8", errors="replace")
in_properties = False
for line in content.splitlines():
if not in_properties:
in_properties = line == "\tProperties: {"
continue
if line == "\t}":
break
key, separator, value = line.partition("=>")
if separator:
self.results[key.strip()] = value.strip()
def _process_sysdiagnose_log(self, file_path: str) -> None:
self.log.info("Found sysdiagnose.log at: %s", file_path)
content = self._get_file_content(file_path).decode("utf-8", errors="replace")
match = re.search(r"sysdiagnose_\S+?\.tar\.gz", content)
if not match:
self.log.info("Could not find the original output path in sysdiagnose.log")
return
file_name = os.path.basename(match.group(0))
try:
created = datetime.strptime(
"_".join(file_name.split("_")[1:3]), "%Y.%m.%d_%H-%M-%S%z"
)
except ValueError:
self.log.warning("Unexpected sysdiagnose file name: %s", file_name)
return
self.results["OriginalFilename"] = file_name
self.results["CreatedTimestamp"] = convert_datetime_to_iso(created)
def run(self) -> None:
for file_path in self._get_files_by_pattern(
"*/logs/appinstallation/appstored.sqlitedb"
):
self._process_appstored(file_path)
for file_path in self._get_files_by_pattern(
"*/logs/MobileActivation/collection_oob_request.txt"
):
self._process_activation_log(file_path)
for file_path in self._get_files_by_pattern("*/remotectl_dumpstate.txt"):
self._process_dumpstate(file_path)
for file_path in self._get_files_by_pattern("*/sysdiagnose.log"):
self._process_sysdiagnose_log(file_path)
# The activation request names the product "iPhone OS"; the model
# description is what an analyst wants to read.
product_name = get_device_desc_from_id(self.results.get("ProductType", ""))
if product_name:
self.results["ProductName"] = product_name
for field in LOGGED_FIELDS:
if field not in self.results:
continue
value = self.results[field]
if field == "BuildVersion" and value:
self.log.info("%s: %s - %s", field, value, find_version_by_build(value))
else:
self.log.info("%s: %s", field, value)
if self.results.get("BuildVersion"):
is_ios_version_outdated(self.results["BuildVersion"], self.log)
@@ -35,16 +35,11 @@ class TestDumpsysAccessibilityArtifact:
assert len(da.results) == 0
da.parse(data)
# One named service, plus one count-only record: the dump states
# `installedServiceCount=2` and names only one component.
assert len(da.results) == 2
assert len(da.results) == 1
assert da.results[0]["package_name"] == "com.malware.accessibility"
assert da.results[0]["service_name"] == "com.malware.service.malwareservice"
assert da.results[0]["enabled"] is True
# This fixture never prints an `installed services:` section, so the
# dump does not state the installed status. Reporting False would turn
# "not stated" into "not installed", so it reads None here.
assert da.results[0]["installed"] is None
assert da.results[0]["installed"] is False
def test_accessibility_service_alert(self):
da = DumpsysAccessibilityArtifact()
@@ -55,11 +50,9 @@ class TestDumpsysAccessibilityArtifact:
da.check_indicators()
assert len(da.alertstore.alerts) == 2
assert len(da.alertstore.alerts) == 1
assert da.alertstore.alerts[0].level == AlertLevel.MEDIUM
assert da.alertstore.alerts[0].event == da.results[0]
assert da.alertstore.alerts[1].level == AlertLevel.LOW
assert da.alertstore.alerts[1].event == da.results[1]
def test_same_component_is_kept_for_each_user(self):
da = DumpsysAccessibilityArtifact()
@@ -91,11 +84,7 @@ User state[attributes:{id=10
assert len(da.alertstore.alerts) == 0
da.check_indicators()
assert len(da.alertstore.alerts) == len(da.results)
# Every service in this fixture is installed and switched off
# (`enabled services:{}` is printed and empty), so the three non-IOC
# findings are LOW, not MEDIUM. The IOC match is unaffected by the
# state.
assert da.alertstore.count(AlertLevel.LOW) == 3
assert da.alertstore.count(AlertLevel.MEDIUM) == 3
assert da.alertstore.count(AlertLevel.CRITICAL) == 1
critical_alert = next(
alert
@@ -1,149 +0,0 @@
# 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/
"""The dump's own installed-service count must survive into the artifact.
Most builds never print the `installed services: {…}` block; they state `installedServiceCount=N` in the user's `attributes:{…}` line and
list nothing. Dropping that number makes an artifact that says "no
accessibility services" about a dump that said there are five.
"""
from mvt.android.artifacts.dumpsys_accessibility import DumpsysAccessibilityArtifact
from mvt.common.alerts import AlertLevel
from ..utils import get_artifact
AOSP_NO_LIST = """\
ACCESSIBILITY MANAGER (dumpsys accessibility)
User state[
attributes:{id=0, touchExplorationEnabled=false, installedServiceCount=5}
Bound services:{}
Enabled services:{}
Binding services:{}
Crashed services:{}
"""
ONE_UI_WITH_LIST = """\
ACCESSIBILITY MANAGER (dumpsys accessibility)
User state[attributes:{id=0, installedServiceCount=2}
installed services: {
0 : com.example.app/com.example.app.Service
1 : com.other.app/.Helper
}
enabled services: {
}
"""
TWO_USERS = """\
ACCESSIBILITY MANAGER (dumpsys accessibility)
User state[attributes:{id=0, installedServiceCount=1}
installed services: {
0 : com.example.app/com.example.app.Service
}
User state[attributes:{id=95, installedServiceCount=3}
Enabled services:{}
"""
PARTIAL_TWO_USERS = """\
ACCESSIBILITY MANAGER (dumpsys accessibility)
User state[attributes:{id=0, installedServiceCount=3}
Enabled services:{{com.example.app/com.example.app.Service}}
User state[attributes:{id=10, installedServiceCount=1}
Enabled services:{{com.other.app/.Helper}}
"""
ZERO_COUNT = """\
ACCESSIBILITY MANAGER (dumpsys accessibility)
User state[attributes:{id=0, installedServiceCount=0}
Enabled services:{}
"""
def _parse(content):
artifact = DumpsysAccessibilityArtifact()
artifact.results = []
artifact.parse(content)
return artifact
class TestAccessibilityInstalledServiceCount:
def test_stated_count_without_a_list_is_kept(self):
artifact = _parse(AOSP_NO_LIST)
assert len(artifact.results) == 1
record = artifact.results[0]
assert record["installed_service_count"] == 5
assert record["component"] is None
# Every state flag stays unknown: the dump named no service to which a
# state could belong.
assert record["installed"] is None
assert record["enabled"] is None
def test_a_stated_count_is_reported_as_low(self):
artifact = _parse(AOSP_NO_LIST)
artifact.check_indicators()
alerts = artifact.alertstore.alerts
assert len(alerts) == 1
# A count is a coverage statement, not a running service: it must not
# compete with a service the dump says is enabled.
assert alerts[0].level == AlertLevel.LOW
assert "does not list their component names" in alerts[0].message
assert "5 installed" in alerts[0].message
def test_a_listed_user_carries_the_count_on_each_service(self):
artifact = _parse(ONE_UI_WITH_LIST)
assert len(artifact.results) == 2
assert {record["installed_service_count"] for record in artifact.results} == {2}
assert all(record["component"] for record in artifact.results)
def test_only_the_unlisted_user_gets_a_count_record(self):
artifact = _parse(TWO_USERS)
listed = [record for record in artifact.results if record["component"]]
unlisted = [record for record in artifact.results if not record["component"]]
assert [record["user_id"] for record in listed] == [0]
assert [record["user_id"] for record in unlisted] == [95]
assert unlisted[0]["installed_service_count"] == 3
def test_a_zero_count_adds_nothing(self):
# "Zero installed" is a negative result the empty section already
# states; a record for it would be noise.
assert _parse(ZERO_COUNT).results == []
def test_a_partly_named_count_reports_the_unnamed_rest(self):
# The Android 14 fixture states `installedServiceCount=2` and names one
# enabled component. The listing is incomplete, and must not read as
# complete.
artifact = DumpsysAccessibilityArtifact()
artifact.results = []
with open(
get_artifact("android_data/dumpsys_accessibility_v14_or_later.txt")
) as handle:
artifact.parse(handle.read())
unlisted = [record for record in artifact.results if not record["component"]]
assert len(unlisted) == 1
assert unlisted[0]["installed_service_count"] == 2
assert unlisted[0]["unnamed_service_count"] == 1
artifact.check_indicators()
low = [
alert
for alert in artifact.alertstore.alerts
if alert.level == AlertLevel.LOW
]
assert len(low) == 1
assert "only 1 of them (1 unnamed)" in low[0].message
def test_the_unnamed_rest_is_counted_per_user(self):
# User 0 names one of three, user 10 names its only one: the gap
# belongs to user 0 alone.
artifact = _parse(PARTIAL_TWO_USERS)
unlisted = [record for record in artifact.results if not record["component"]]
assert [
(record["user_id"], record["unnamed_service_count"]) for record in unlisted
] == [(0, 2)]
def test_a_fully_named_count_adds_nothing(self):
artifact = _parse(ONE_UI_WITH_LIST)
assert all(
record["unnamed_service_count"] is None for record in artifact.results
)
@@ -1,155 +0,0 @@
# 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/
"""Installed is not the same as enabled, and the dump says which.
Two things this guards:
* a section the dump never printed must read as None ("not stated"), not as
False ("not enabled");
* the alert must name the state, instead of firing identically on a device
where nothing is switched on and one where something is bound.
"""
from types import SimpleNamespace
from mvt.android.artifacts.dumpsys_accessibility import DumpsysAccessibilityArtifact
from mvt.common.alerts import AlertLevel
from ..utils import get_artifact
NO_STATE_SECTIONS = """\
ACCESSIBILITY MANAGER (dumpsys accessibility)
User state[attributes:{id=0, currentUser=true}
installed services: {
0 : com.example.app/com.example.app.Service
}
"""
ENABLED_BLOCK = """\
ACCESSIBILITY MANAGER (dumpsys accessibility)
User state[attributes:{id=0, currentUser=true}
installed services: {
0 : com.example.app/com.example.app.Service
1 : com.other.app/.Helper
}
enabled services: {
0 : com.other.app/.Helper
}
bound services:{
0 : com.other.app/.Helper
}
"""
# User 0 prints only `installed services`, user 10 only `Enabled services`.
# Neither section speaks for the other user.
TWO_USERS_DIFFERENT_SECTIONS = """\
ACCESSIBILITY MANAGER (dumpsys accessibility)
User state[attributes:{id=0, currentUser=true}
installed services: {
0 : com.example.app/com.example.app.Service
}
User state[attributes:{id=10, currentUser=false}
Enabled services:{{com.other.app/.Helper}}
"""
class _IndicatorsMatching:
"""Minimal stand-in: matches one package id, like the STIX2 loader would."""
def __init__(self, package_name: str) -> None:
self.package_name = package_name
def check_app_id(self, app_id):
if app_id != self.package_name:
return None
return SimpleNamespace(
message=f"Found a known suspicious app: {app_id}", ioc={"value": app_id}
)
class TestAccessibilityServiceState:
def _parse(self, content):
artifact = DumpsysAccessibilityArtifact()
artifact.results = []
artifact.parse(content)
return {r["component"]: r for r in artifact.results}
def test_absent_sections_leave_the_state_unknown(self):
# None, not False: a build that does not print the sections says
# nothing about what is enabled, and that must not read as "nothing".
state = self._parse(NO_STATE_SECTIONS)[
"com.example.app/com.example.app.Service"
]
assert state["installed"] is True
assert state["enabled"] is None
assert state["bound"] is None
def test_enabled_and_bound_are_attributed_per_service(self):
results = self._parse(ENABLED_BLOCK)
installed_only = results["com.example.app/com.example.app.Service"]
active = results["com.other.app/.Helper"]
assert (installed_only["enabled"], installed_only["bound"]) == (False, False)
assert (active["enabled"], active["bound"]) == (True, True)
def test_alert_message_carries_the_state(self):
artifact = DumpsysAccessibilityArtifact()
artifact.results = []
with open(get_artifact("android_data/dumpsys_accessibility.txt")) as handle:
artifact.parse(handle.read())
artifact.check_indicators()
assert artifact.alertstore.alerts
assert all("installed" in alert.message for alert in artifact.alertstore.alerts)
def test_a_switched_off_service_is_low_and_a_running_one_medium(self):
# A service the dump says is OFF still reaches the analyst, but must not
# compete with one that is actually bound. "Not stated" is not "off".
artifact = DumpsysAccessibilityArtifact()
artifact.results = []
artifact.parse(ENABLED_BLOCK)
artifact.check_indicators()
by_level = {}
for alert in artifact.alertstore.alerts:
by_level.setdefault(alert.level, []).append(alert.message)
assert artifact.alertstore.count(AlertLevel.LOW) == 1
assert artifact.alertstore.count(AlertLevel.MEDIUM) == 1
assert "com.example.app" in by_level[AlertLevel.LOW][0]
assert "com.other.app" in by_level[AlertLevel.MEDIUM][0]
def test_an_unstated_enabled_state_stays_medium(self):
artifact = DumpsysAccessibilityArtifact()
artifact.results = []
artifact.parse(NO_STATE_SECTIONS)
artifact.check_indicators()
assert artifact.alertstore.count(AlertLevel.MEDIUM) == 1
assert artifact.alertstore.count(AlertLevel.LOW) == 0
def test_a_disabled_service_is_still_matched_against_indicators(self):
# The state decides the severity of an ordinary finding, never whether
# the package is compared with the IOC feeds.
artifact = DumpsysAccessibilityArtifact()
artifact.results = []
artifact.parse(ENABLED_BLOCK)
artifact.indicators = _IndicatorsMatching("com.example.app")
artifact.check_indicators()
assert artifact.alertstore.count(AlertLevel.CRITICAL) == 1
assert artifact.alertstore.count(AlertLevel.LOW) == 0
def test_printed_sections_are_tracked_per_user(self):
artifact = DumpsysAccessibilityArtifact()
artifact.results = []
artifact.parse(TWO_USERS_DIFFERENT_SECTIONS)
by_user = {r["user_id"]: r for r in artifact.results}
# User 0's enabled state is not stated, so it is unknown, not off.
assert (by_user[0]["installed"], by_user[0]["enabled"]) == (True, None)
# User 10's installed state is not stated either.
assert (by_user[10]["installed"], by_user[10]["enabled"]) == (None, True)
artifact.check_indicators()
assert artifact.alertstore.count(AlertLevel.LOW) == 0
assert artifact.alertstore.count(AlertLevel.MEDIUM) == 2
assert not any(
"installed, not enabled" in alert.message
for alert in artifact.alertstore.alerts
)
-43
View File
@@ -1,43 +0,0 @@
# 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/
"""A mount is writable only if both mountinfo option layers allow it."""
from mvt.android.artifacts.mounts import Mounts
from mvt.common.alerts import AlertLevel
# fields[5] is the per-mount (VFS) layer, the field after "-" the superblock.
RO_OVERLAY = (
"304 166 0:130 / /product/usr rw,relatime shared:63 - overlay overlay "
"ro,seclabel,lowerdir=/mnt/vendor/ext/product/usr:/product/usr"
)
RW_SYSTEM = (
"305 166 0:131 / /system rw,relatime shared:64 - ext4 /dev/block/dm-1 "
"rw,seclabel,errors=panic"
)
def _run(line):
mounts = Mounts()
mounts.results = Mounts.parse_mountinfo(line, 1)
mounts.check_indicators()
return mounts
class TestMountsReadWriteLayers:
def test_rw_vfs_over_ro_superblock_is_not_writable(self):
mounts = _run(RO_OVERLAY)
assert mounts.results[0]["is_read_write"] is False
assert mounts.alertstore.count(AlertLevel.HIGH) == 0
assert mounts.alertstore.count(AlertLevel.MEDIUM) == 0
def test_rw_in_both_layers_still_alerts(self):
mounts = _run(RW_SYSTEM)
assert mounts.results[0]["is_read_write"] is True
assert mounts.alertstore.count(AlertLevel.HIGH) == 1
def test_both_option_layers_are_still_reported(self):
mounts = _run(RO_OVERLAY)
options = mounts.results[0]["options_list"]
assert "rw" in options and "ro" in options
@@ -1,42 +0,0 @@
# 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/
"""A section ends at the next section, not at a timing line printed inside it."""
from mvt.android.modules.bugreport.base import BugReportModule
# dumpstate prints a section's duration when that section finishes, which can
# land in the middle of the section currently being written.
DUMPSTATE = """\
------ SYSTEM PROPERTIES (getprop) ------
[nfc.initialized]: [true]
------ 0.101s was the duration of 'DROPBOX SYSTEM SERVER CRASHES' ------
[ro.build.version.sdk]: [30]
[ro.product.model]: [SM-A305F]
------ 0.064s was the duration of 'SYSTEM PROPERTIES' ------
------ STORAGE INFO (df) ------
/dev/root 2.9G
"""
class TestExtractCommandSection:
def test_a_foreign_timing_line_does_not_end_the_section(self):
section = BugReportModule.extract_command_section(
DUMPSTATE, "------ SYSTEM PROPERTIES"
)
assert "[ro.product.model]: [SM-A305F]" in section
assert section.count("\n") == 2
def test_the_next_section_is_still_the_boundary(self):
section = BugReportModule.extract_command_section(
DUMPSTATE, "------ SYSTEM PROPERTIES"
)
assert "STORAGE INFO" not in section
assert "/dev/root" not in section
def test_timing_lines_are_not_returned_as_content(self):
section = BugReportModule.extract_command_section(
DUMPSTATE, "------ SYSTEM PROPERTIES"
)
assert "was the duration of" not in section
@@ -1,56 +0,0 @@
# Mobile Verification Toolkit (MVT)
# Copyright (c) 2021-2023 The MVT Authors.
# Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/
"""An OEM wrapper archive must not read as an empty bug report.
MIUI / HyperOS hands out a zip of app logs with the real
`bugreport-<device>-<timestamp>.zip` nested inside. The outer archive has
none of the entry points the modules read, so every module
reported it found nothing and the command still exited 0 — an empty analysis
that looks like a finished one.
"""
import io
import zipfile
from mvt.android.cmd_check_bugreport import CmdAndroidCheckBugreport
DUMPSTATE = "== dumpstate: 2026-01-01 00:00:00\nDUMP OF SERVICE package:\n"
def _inner_zip() -> bytes:
buffer = io.BytesIO()
with zipfile.ZipFile(buffer, "w") as inner:
inner.writestr("main_entry.txt", "bugreport-test-2026-01-01-00-00-00.txt")
inner.writestr("bugreport-test-2026-01-01-00-00-00.txt", DUMPSTATE)
return buffer.getvalue()
def _wrapper_zip() -> zipfile.ZipFile:
buffer = io.BytesIO()
with zipfile.ZipFile(buffer, "w") as outer:
outer.writestr("app_logs/hilog.txt", "unrelated OEM log\n")
outer.writestr("bugreport-test-2026-01-01-00-00-00.zip", _inner_zip())
return zipfile.ZipFile(io.BytesIO(buffer.getvalue()))
class TestCheckBugreportWrapper:
def test_nested_bugreport_is_used(self, tmp_path):
cmd = CmdAndroidCheckBugreport(results_path=str(tmp_path))
cmd.from_zip(_wrapper_zip())
module = cmd.modules[0](results_path=str(tmp_path))
cmd.module_init(module)
assert "main_entry.txt" in module.zip_files
def test_plain_bugreport_is_left_alone(self, tmp_path):
buffer = io.BytesIO()
with zipfile.ZipFile(buffer, "w") as archive:
archive.writestr("main_entry.txt", "bugreport.txt")
archive.writestr("bugreport.txt", DUMPSTATE)
archive.writestr("attachments/extra.zip", _inner_zip())
cmd = CmdAndroidCheckBugreport(results_path=str(tmp_path))
cmd.from_zip(zipfile.ZipFile(io.BytesIO(buffer.getvalue())))
module = cmd.modules[0](results_path=str(tmp_path))
cmd.module_init(module)
assert "attachments/extra.zip" in module.zip_files
-85
View File
@@ -5,13 +5,8 @@
import json
import logging
from unittest.mock import patch
import pytest
from click.testing import CliRunner
from mvt.common.command import Command
from mvt.common.indicators import Indicators
from mvt.common.module import MVTModule
@@ -202,86 +197,6 @@ class RecordingCommand(Command):
class TestCommand:
def test_listing_modules_does_not_load_indicators(self):
with patch("mvt.common.command.Indicators.load_indicators_files") as load:
cmd = RecordingCommand()
cmd.list_modules()
load.assert_not_called()
def test_output_folder_is_created_by_a_run_not_by_listing(self, tmp_path):
output_path = tmp_path / "out"
with patch("mvt.common.command.Indicators.load_indicators_files"):
cmd = RecordingCommand(results_path=str(output_path))
cmd.list_modules()
assert not output_path.exists()
cmd.run()
assert (output_path / "command.log").is_file()
def test_indicators_load_once_and_are_shared(self, indicator_file, monkeypatch):
from mvt.common.config import settings
monkeypatch.setattr(settings, "STIX2", "")
monkeypatch.setattr(Indicators, "_load_downloaded_indicators", lambda self: None)
original = Indicators.load_indicators_files
with patch.object(
Indicators, "load_indicators_files", autospec=True, side_effect=original
) as load:
cmd = RecordingCommand(ioc_files=[indicator_file])
load.assert_not_called()
indicators = cmd.iocs
assert indicators.total_ioc_count == 9
assert len(indicators.ioc_collections) == 1
assert cmd.iocs is indicators
child = RecordingCommand(iocs=indicators)
assert child.iocs is indicators
load.assert_called_once_with(indicators, [indicator_file])
@pytest.mark.parametrize("assign", [False, True])
def test_supplied_empty_indicators_are_not_loaded(self, assign):
indicators = Indicators(logging.getLogger(__name__))
with patch.object(Indicators, "load_indicators_files") as load:
cmd = RecordingCommand(iocs=None if assign else indicators)
if assign:
cmd.iocs = indicators
assert cmd.iocs is indicators
assert cmd.iocs.total_ioc_count == 0
load.assert_not_called()
@pytest.mark.parametrize("list_modules", [False, True])
def test_backup_cli_does_not_load_indicators_before_analysis(
self, tmp_path, list_modules
):
from mvt.ios.cli import check_backup
args = [str(tmp_path)]
if list_modules:
args.insert(0, "--list-modules")
with patch.object(Indicators, "load_indicators_files") as load:
result = CliRunner().invoke(check_backup, args)
assert result.exit_code == (0 if list_modules else 1)
load.assert_not_called()
def test_run_checks_synthetic_indicators(self, indicator_file, monkeypatch):
from mvt.common.config import settings
monkeypatch.setattr(settings, "STIX2", "")
monkeypatch.setattr(Indicators, "_load_downloaded_indicators", lambda self: None)
class MatchingModule(RecordingModule):
def run(self):
self.results = ["https://example.org/test"]
def check_indicators(self):
self.detected = [
url for url in self.results if self.indicators.check_domain(url)
]
cmd = RecordingCommand(ioc_files=[indicator_file])
cmd.modules = [MatchingModule]
cmd.run()
assert cmd.executed[0].detected == ["https://example.org/test"]
assert cmd.executed[0].indicators is cmd.iocs
def setup_method(self):
RecordingModule.run_order = []
+1 -4
View File
@@ -12,16 +12,13 @@ from mvt.ios.command_modules import IOS_CHECK_IOCS_MODULES
from mvt.ios.modules.backup import BACKUP_MODULES as IOS_BACKUP_MODULES
from mvt.ios.modules.fs import FS_MODULES
from mvt.ios.modules.mixed import MIXED_MODULES
from mvt.ios.modules.sysdiagnose import SYSDIAGNOSE_MODULES
def test_the_check_iocs_lists_are_the_families_of_their_platform():
# The CLI reads these same lists, so nothing composing one elsewhere can
# drift from what the command runs. This pins what the lists are composed
# of.
assert IOS_CHECK_IOCS_MODULES == (
IOS_BACKUP_MODULES + FS_MODULES + MIXED_MODULES + SYSDIAGNOSE_MODULES
)
assert IOS_CHECK_IOCS_MODULES == IOS_BACKUP_MODULES + FS_MODULES + MIXED_MODULES
assert ANDROID_CHECK_IOCS_MODULES == (
ANDROID_BACKUP_MODULES
+ BUGREPORT_MODULES
-28
View File
@@ -311,31 +311,3 @@ class TestIndicators:
ind = Indicators(log=logging)
ind.load_indicators_files([], load_default=False)
assert ind.total_ioc_count == 9
def test_check_url_matches_w_prefixed_subdomain(self, tmp_path):
import json
stix_file = tmp_path / "w-domain.stix2"
stix_file.write_text(
json.dumps(
{
"objects": [
{
"type": "indicator",
"pattern": "[domain-name:value = 'web.evil.com']",
}
]
}
),
encoding="utf-8",
)
ind = Indicators(log=logging)
ind.load_indicators_files([str(stix_file)], load_default=False)
for url in (
"https://web.evil.com/path",
"https://www.web.evil.com/path",
):
match = ind.check_url(url)
assert match is not None
assert match.ioc.value == "web.evil.com"
-20
View File
@@ -22,23 +22,3 @@ def test_google_maps_url_is_not_shortened(url):
def test_other_google_short_url_is_shortened():
assert URL("https://goo.gl/example").check_if_shortened() is True
@pytest.mark.parametrize(
"url, domain",
[
("https://www.example.com/path", "example.com"),
# Only the whole "www." prefix comes off, not any leading "w" or "." character.
("https://web.example.com", "web.example.com"),
("https://wow.com", "wow.com"),
("https://wired.com", "wired.com"),
("https://www.wow.com", "wow.com"),
],
)
def test_get_domain_strips_only_a_whole_www_prefix(url, domain):
assert URL(url).domain == domain
def test_shortener_starting_with_w_is_detected():
assert URL("https://w3t.org/example").check_if_shortened() is True
assert URL("https://www.w3t.org/example").check_if_shortened() is True
+13
View File
@@ -18,6 +18,7 @@ from mvt.common.utils import (
generate_hashes_from_path,
get_sha256_from_file_path,
init_logging,
sanitize_json_data,
set_verbose_logging,
)
@@ -108,6 +109,18 @@ class TestCustomJSONEncoder:
)
def test_sanitize_json_data_preserves_nested_binary_and_dates():
value = {
"blob": b"\x00\xff",
"nested": [datetime(2023, 11, 13, 12, 21, 49, 727467)],
}
assert sanitize_json_data(value) == {
"blob": "AP8=",
"nested": ["2023-11-13 12:21:49.727467"],
}
class TestInitLogging:
def test__init_logging_is_idempotent(self):
# Loaded module packages may import an MVT CLI module, which calls
+7 -20
View File
@@ -3,25 +3,19 @@
# Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/
import atexit
import logging
import os
import shutil
import tempfile
import pytest
from .artifacts.generate_stix import generate_test_stix_file
from mvt.common.cli_plugins import (
MVT_ANDROID_CUSTOM_COMMANDS_ENV,
MVT_CUSTOM_COMMANDS_ENV,
MVT_IOS_CUSTOM_COMMANDS_ENV,
)
from mvt.common.indicators import Indicators
# The suite must neither read nor write the developer's own MVT settings,
# downloaded indicators or update-check state, and mvt.common.config saves
# the settings file as soon as it is imported. Both folders are redirected
# before any mvt module is imported, which is why this file imports none at
# the top; the subprocesses the tests start inherit the variables.
MVT_TEST_HOME = tempfile.mkdtemp(prefix="mvt-tests-")
atexit.register(shutil.rmtree, MVT_TEST_HOME, ignore_errors=True)
os.environ["MVT_CONFIG_FOLDER"] = os.path.join(MVT_TEST_HOME, "config")
os.environ["MVT_DATA_FOLDER"] = os.path.join(MVT_TEST_HOME, "data")
from .artifacts.generate_stix import generate_test_stix_file
@pytest.fixture(scope="session", autouse=True)
@@ -53,8 +47,6 @@ def indicators_factory(indicator_file):
android_property_names=[],
files_sha256=[],
):
from mvt.common.indicators import Indicators
ind = Indicators(log=logging.getLogger())
ind.parse_stix2(indicator_file)
@@ -85,11 +77,6 @@ def restore_cli_commands(monkeypatch):
"""
from mvt.android.cli import cli as android_cli
from mvt.cli import cli as neutral_cli
from mvt.common.cli_plugins import (
MVT_ANDROID_CUSTOM_COMMANDS_ENV,
MVT_CUSTOM_COMMANDS_ENV,
MVT_IOS_CUSTOM_COMMANDS_ENV,
)
from mvt.ios.cli import cli as ios_cli
groups = (neutral_cli, ios_cli, android_cli)
+2
View File
@@ -21,6 +21,8 @@ class TestCalendarModule:
assert len(m.timeline) == 4
assert len(m.alertstore.alerts) == 0
assert m.results[0]["summary"] == "Super interesting meeting"
assert m.results[0]["record"]["calendar_app_link"] is None
assert "participant_proposed_start_date" in m.results[0]["record"]
def test_calendar_with_explicit_file_path(self):
backup_path = get_ios_backup_folder()
+38
View File
@@ -0,0 +1,38 @@
# 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 sqlite3
from mvt.common.module import run_module
from mvt.ios.modules.mixed.calls import Calls
def test_calls_preserves_complete_source_record(tmp_path):
db_path = tmp_path / "CallHistory.storedata"
conn = sqlite3.connect(db_path)
conn.execute(
"""
CREATE TABLE ZCALLRECORD (
ZDATE REAL, ZDURATION REAL, ZLOCATION TEXT, ZADDRESS BLOB,
ZSERVICE_PROVIDER TEXT, ZORIGINATED INTEGER, ZANSWERED INTEGER,
ZMETADATA BLOB
);
"""
)
conn.execute(
"INSERT INTO ZCALLRECORD VALUES (?, ?, ?, ?, ?, ?, ?, ?);",
(700000000, 42, "Berlin", b"+491234", "carrier", 1, 0, b"\x00\xff"),
)
conn.commit()
conn.close()
module = Calls(file_path=str(db_path))
run_module(module)
assert len(module.results) == 1
assert module.results[0]["number"] == "+491234"
assert module.results[0]["call"]["ZORIGINATED"] == 1
assert module.results[0]["call"]["ZANSWERED"] == 0
assert module.results[0]["call"]["ZMETADATA"] == "AP8="
+42
View File
@@ -0,0 +1,42 @@
# 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 sqlite3
from mvt.common.module import run_module
from mvt.ios.modules.mixed.contacts import Contacts
def test_contacts_preserves_complete_rows_and_people_without_values(tmp_path):
db_path = tmp_path / "AddressBook.sqlitedb"
conn = sqlite3.connect(db_path)
conn.executescript(
"""
CREATE TABLE ABPerson (
ROWID INTEGER PRIMARY KEY, First TEXT, Middle TEXT, Last TEXT,
Organization TEXT, Note TEXT
);
CREATE TABLE ABMultiValue (
ROWID INTEGER PRIMARY KEY, record_id INTEGER, value TEXT,
label INTEGER
);
INSERT INTO ABPerson VALUES (1, 'Alice', NULL, 'Example', NULL, 'note');
INSERT INTO ABPerson VALUES (2, 'Bob', NULL, 'NoValue', NULL, 'retained');
INSERT INTO ABMultiValue VALUES (10, 1, '+491234', 3);
"""
)
conn.close()
module = Contacts(file_path=str(db_path))
run_module(module)
assert len(module.results) == 2
alice = next(result for result in module.results if result["first"] == "Alice")
bob = next(result for result in module.results if result["first"] == "Bob")
assert alice["contact"]["person_Note"] == "note"
assert alice["contact"]["multivalue_label"] == 3
assert alice["First"] == "Alice"
assert bob["value"] is None
assert bob["contact"]["person_Note"] == "retained"
+2
View File
@@ -20,6 +20,8 @@ class TestDatausageModule:
assert m.results[0]["isodate"][0:19] == "2019-08-27 15:08:09"
assert len(m.results) == 42
assert len(m.timeline) == 60
assert "live_ZBILLCYCLEEND" in m.results[0]["record"]
assert "process_Z_ENT" in m.results[0]["record"]
assert (
len(m.alertstore.alerts) == 1
) # We now have a detection for missing processes.
+4
View File
@@ -20,6 +20,10 @@ class TestInteractionCModule:
run_module(m)
assert len(m.results) == 3
assert all(
result["interaction"]["Z_PK"] == result["table_id"] for result in m.results
)
assert all("ZMECHANISM" in result["interaction"] for result in m.results)
incoming = next(
r for r in m.results if r["sender_identifier"] == "100000000000001@lid"
@@ -5,6 +5,7 @@
import logging
import shutil
import sqlite3
import pytest
@@ -52,6 +53,7 @@ class TestSafariBrowserStateModule:
assert len(m.results) == 1
assert len(m.timeline) == 1
assert len(m.alertstore.alerts) == 0
assert "tab" in m.results[0]
def test_parsing_backup_with_profile(self, backup_with_safari_profile):
m = SafariBrowserState(target_path=backup_with_safari_profile)
@@ -74,3 +76,30 @@ class TestSafariBrowserStateModule:
assert len(m.alertstore.alerts) == 1
assert len(m.results) == 1
assert m.results[0]["tab_url"] == "https://en.wikipedia.org/wiki/NSO_Group"
def test_tabs_without_session_rows_are_not_dropped(self, tmp_path):
db_path = tmp_path / "BrowserState.db"
conn = sqlite3.connect(db_path)
conn.executescript(
"""
CREATE TABLE tabs (
uuid TEXT, title TEXT, url TEXT, user_visible_url TEXT,
last_viewed_time REAL, private_browsing INTEGER
);
CREATE TABLE tab_sessions (tab_uuid TEXT, session_data BLOB);
INSERT INTO tabs VALUES (
'tab-1', 'Example', 'https://example.test',
'https://example.test', 700000000, 1
);
"""
)
conn.close()
module = SafariBrowserState(target_path=str(tmp_path))
module._process_browser_state_db(str(db_path))
assert len(module.results) == 1
assert module.results[0]["tab_url"] == "https://example.test"
assert module.results[0]["session_data"] == []
assert module.results[0]["tab"]["private_browsing"] == 1
assert module.results[0]["tab"]["session_data"] is None
+1
View File
@@ -105,6 +105,7 @@ class TestSafariHistoryModule:
assert len(m.results) == 2
assert {result["url"] for result in m.results} == {DEFAULT_URL, PROFILE_URL}
assert len({result["safari_history_db"] for result in m.results}) == 2
assert all("visit_history_item" in result["record"] for result in m.results)
def test_parsing_fs_dump_with_profile(self, fs_dump_with_safari_profile):
m = SafariHistory(target_path=fs_dump_with_safari_profile)
+3
View File
@@ -22,6 +22,9 @@ class TestTCCModule:
assert m.results[0]["service"] == "kTCCServiceUbiquity"
assert m.results[0]["client"] == "com.apple.Preferences"
assert m.results[0]["auth_value"] == "allowed"
assert m.results[0]["auth_value_value"] == 2
assert "policy_id" in m.results[0]
assert "csreq" in m.results[0]
def test_tcc_detection(self, indicator_file):
m = TCC(target_path=get_ios_backup_folder())
@@ -34,6 +34,7 @@ class TestWebkitResourceLoadStatisticsModule:
"most_recent_web_push_interaction_time" not in result
for result in m.results
)
assert all("record" in result for result in m.results)
def test_webkit_full_timestamp_schema(self, tmp_path):
db_path = tmp_path / "observations.db"
@@ -68,3 +69,4 @@ class TestWebkitResourceLoadStatisticsModule:
assert "most_recent_user_interaction_time_isodate" in result
assert result["most_recent_web_push_interaction_time"] == -1.0
assert "most_recent_web_push_interaction_time_isodate" not in result
assert result["record"]["domainID"] == 1
-4
View File
@@ -1,4 +0,0 @@
# 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/
@@ -1,161 +0,0 @@
# 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 json
import plistlib
import sqlite3
import tarfile
from mvt.common.module import run_module
from mvt.ios.cmd_check_sysdiagnose import CmdIOSCheckSysdiagnose
from mvt.ios.modules.sysdiagnose.sysdiagnose_info import SysdiagnoseInfo
from mvt.ios.versions import get_device_desc_from_id
# The name sysdiagnose gives its archive: the time it ran, then the OS and build.
ARCHIVE_NAME = "sysdiagnose_2024.01.02_03-04-05+0200_iPhone-OS_iPhone_21C62"
DUMPSTATE = (
"Found device: ...\n"
"\tProperties: {\n"
"\t\tProductType => iPhone12,1\n"
"\t\tOSVersion => 17.2\n"
"\t\tSerialNumber => C0FFEE000000\n"
"\t\tRegionCode => LL\n"
"\t}\n"
"\tServices: {\n"
"\t\tcom.apple.example => ignored\n"
"\t}\n"
)
ACTIVATION_BODY = {
"serial-number": "C0FFEE000000",
"productType": "iPhone12,1",
"productName": "iPhone OS",
"imei": "000000000000000",
"os-version": "17.2",
"os-build": "21C62",
"udid": "00000000-0000000000000000",
"meid": "00000000000000",
}
def make_sysdiagnose(tmp_path, activation_body=None):
folder = tmp_path / ARCHIVE_NAME
folder.mkdir()
(folder / "sysdiagnose.log").write_text(
f"Output available at '/private/var/tmp/{ARCHIVE_NAME}.tar.gz'\n",
encoding="utf-8",
)
(folder / "remotectl_dumpstate.txt").write_text(DUMPSTATE, encoding="utf-8")
activation = folder / "logs" / "MobileActivation"
activation.mkdir(parents=True)
body = json.dumps(
activation_body if activation_body is not None else ACTIVATION_BODY
)
(activation / "collection_oob_request.txt").write_text(
f"HEADERS: {{}}\nBODY: {body}\nEND\n", encoding="utf-8"
)
appinstallation = folder / "logs" / "appinstallation"
appinstallation.mkdir(parents=True)
conn = sqlite3.connect(appinstallation / "appstored.sqlitedb")
conn.execute("CREATE TABLE asset (sinfs_data BLOB)")
conn.execute(
"INSERT INTO asset VALUES (?)",
(plistlib.dumps([{"sinf": b"\x00\x10nameExample Person\x00\x00rest"}]),),
)
conn.execute("CREATE TABLE job_software (store_account_name TEXT)")
conn.execute("INSERT INTO job_software VALUES (NULL)")
conn.execute("INSERT INTO job_software VALUES ('person@example.com')")
conn.commit()
conn.close()
return folder
def run_command(target, results_path=None):
command = CmdIOSCheckSysdiagnose(target_path=str(target), results_path=results_path)
command.run()
(module,) = [m for m in command.executed if isinstance(m, SysdiagnoseInfo)]
return module
def test_device_details_from_a_sysdiagnose_folder(tmp_path):
results_path = tmp_path / "results"
results_path.mkdir()
module = run_command(make_sysdiagnose(tmp_path), str(results_path))
assert module.results["SerialNumber"] == "C0FFEE000000"
assert module.results["ProductType"] == "iPhone12,1"
assert module.results["ProductName"] == get_device_desc_from_id("iPhone12,1")
assert module.results["ProductName"] != "iPhone OS"
assert module.results["OSVersion"] == "17.2"
assert module.results["BuildVersion"] == "21C62"
assert module.results["UniqueIdentifier"] == "00000000-0000000000000000"
assert module.results["RegionCode"] == "LL"
assert "com.apple.example" not in module.results
assert module.results["Account Name"] == "Example Person"
assert module.results["Email Address"] == "person@example.com"
assert module.results["OriginalFilename"] == f"{ARCHIVE_NAME}.tar.gz"
assert module.results["CreatedTimestamp"] == "2024-01-02 01:04:05.000000"
assert (results_path / "sysdiagnose_info.json").exists()
def test_device_details_from_a_sysdiagnose_archive(tmp_path):
folder = make_sysdiagnose(tmp_path)
archive_path = tmp_path / f"{ARCHIVE_NAME}.tar.gz"
with tarfile.open(archive_path, "w:gz") as archive:
archive.add(folder, arcname=ARCHIVE_NAME)
module = run_command(archive_path)
assert module.results["SerialNumber"] == "C0FFEE000000"
assert module.results["Account Name"] == "Example Person"
assert module.results["OriginalFilename"] == f"{ARCHIVE_NAME}.tar.gz"
def test_wal_sidecars_are_copied_beside_the_database(tmp_path):
folder = tmp_path / ARCHIVE_NAME
(folder / "logs").mkdir(parents=True)
(folder / "logs" / "db.sqlite").write_bytes(b"main")
(folder / "logs" / "db.sqlite-wal").write_bytes(b"wal")
module = SysdiagnoseInfo()
module.from_sysdiagnose_folder(
str(folder),
[f"{ARCHIVE_NAME}/logs/db.sqlite", f"{ARCHIVE_NAME}/logs/db.sqlite-wal"],
)
copies = tmp_path / "copies"
copies.mkdir()
db_path = module._copy_sqlite_db(f"{ARCHIVE_NAME}/logs/db.sqlite", str(copies))
assert db_path == str(copies / "db.sqlite")
assert (copies / "db.sqlite").read_bytes() == b"main"
assert (copies / "db.sqlite-wal").read_bytes() == b"wal"
assert not (copies / "db.sqlite-shm").exists()
def test_a_sysdiagnose_without_the_files_yields_nothing(tmp_path):
folder = tmp_path / ARCHIVE_NAME
folder.mkdir()
(folder / "other.txt").write_text("nothing here", encoding="utf-8")
module = SysdiagnoseInfo()
module.from_sysdiagnose_folder(str(folder), [f"{ARCHIVE_NAME}/other.txt"])
run_module(module)
assert module.results == {}
def test_a_malformed_activation_request_is_skipped(tmp_path):
folder = make_sysdiagnose(tmp_path)
(folder / "logs" / "MobileActivation" / "collection_oob_request.txt").write_text(
"BODY: {not json}\n", encoding="utf-8"
)
module = run_command(folder)
assert "IMEI" not in module.results
assert module.results["SerialNumber"] == "C0FFEE000000"
-92
View File
@@ -3,16 +3,11 @@
# Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/
import datetime
import logging
import os
import shutil
import zipfile
from click.testing import CliRunner
from mvt.android.cli import check_bugreport
from mvt.android.cmd_check_bugreport import CmdAndroidCheckBugreport
from .utils import get_artifact_folder
@@ -33,90 +28,3 @@ class TestCheckBugreportCommand:
assert result.exit_code == 1
assert "Invalid bugreport archive" in result.output
assert "Traceback" not in result.output
PROPERTIES = (
"------ SYSTEM PROPERTIES (getprop) ------\n"
"[persist.sys.timezone]: [Africa/Nairobi]\n"
"------ 0.01s was the duration of 'SYSTEM PROPERTIES' ------\n"
)
TOMBSTONE = "android_data/bugreport/FS/data/tombstones/tombstone_00"
def _bugreport_zip(tmp_path, dumpstate=PROPERTIES):
"""A bugreport zip holding one tombstone written at 11:38:10 device time.
An even second: zip entry times have a two-second resolution.
"""
path = tmp_path / "bugreport.zip"
with open(os.path.join(get_artifact_folder(), TOMBSTONE), "rb") as handle:
tombstone = handle.read()
with zipfile.ZipFile(path, "w") as archive:
archive.writestr("main_entry.txt", "dumpstate.txt")
archive.writestr("dumpstate.txt", dumpstate)
entry = zipfile.ZipInfo(
"FS/data/tombstones/tombstone_00", date_time=(2023, 3, 10, 11, 38, 10)
)
archive.writestr(entry, tombstone)
return str(path)
def _tombstone_timestamp(target, **options):
cmd = CmdAndroidCheckBugreport(
target_path=target,
module_name="Tombstones",
disable_version_check=True,
disable_indicator_check=True,
**options,
)
cmd.run()
return cmd, cmd.executed[0].results[0]["file_timestamp"]
class TestCheckBugreportTimezone:
def test_zip_entry_times_are_read_in_the_device_timezone(self, tmp_path):
cmd, file_timestamp = _tombstone_timestamp(_bugreport_zip(tmp_path))
assert cmd.module_options["device_timezone"] == "Africa/Nairobi"
# 11:38:10 in Nairobi is 08:38:10 UTC.
assert file_timestamp == "2023-03-10 08:38:10.000000"
def test_timezone_option_wins_over_the_bugreport(self, tmp_path):
_, file_timestamp = _tombstone_timestamp(
_bugreport_zip(tmp_path), module_options={"device_timezone": "Europe/Paris"}
)
assert file_timestamp == "2023-03-10 10:38:10.000000"
result = CliRunner().invoke(
check_bugreport,
["-t", "Europe/Paris", "-m", "Tombstones", _bugreport_zip(tmp_path)],
)
assert result.exit_code == 0, result.output
def test_without_a_timezone_the_wall_clock_is_kept_and_a_warning_given(
self, tmp_path, caplog
):
with caplog.at_level(logging.WARNING, logger="mvt"):
_, file_timestamp = _tombstone_timestamp(_bugreport_zip(tmp_path, ""))
assert file_timestamp == "2023-03-10 11:38:10.000000"
assert "persist.sys.timezone not found" in caplog.text
def test_unpacked_bugreport_warns_and_reads_mtimes_as_utc(self, tmp_path, caplog):
unpacked = tmp_path / "bugreport"
shutil.copytree(
os.path.join(get_artifact_folder(), "android_data/bugreport"), unpacked
)
instant = datetime.datetime(
2023, 3, 10, 8, 38, 11, tzinfo=datetime.timezone.utc
).timestamp()
for name in ("tombstone_00", "tombstone_01"):
os.utime(unpacked / "FS" / "data" / "tombstones" / name, (instant, instant))
with caplog.at_level(logging.WARNING, logger="mvt"):
_, file_timestamp = _tombstone_timestamp(str(unpacked))
assert "unpacked bugreport" in caplog.text
# Whatever the zone of the machine running the analysis.
assert file_timestamp == "2023-03-10 08:38:11.000000"
-8
View File
@@ -19,14 +19,6 @@ class TestCheckBackupCommand:
result = runner.invoke(check_backup, [path])
assert result.exit_code == 0
def test_check_logs_the_backup_path_to_the_command_log(self, tmp_path):
path = get_ios_backup_folder()
output_path = tmp_path / "out"
result = CliRunner().invoke(check_backup, ["--output", str(output_path), path])
assert result.exit_code == 0
command_log = (output_path / "command.log").read_text(encoding="utf-8")
assert f"Checking iTunes backup located at: {path}" in command_log
def test_check_finds_backup_in_subfolder(self, tmp_path, caplog):
runner = CliRunner()
backup_path = tmp_path / "MobileSync" / "Backup" / "device-id"
+4 -39
View File
@@ -1,7 +1,3 @@
import logging
import os
import tarfile
from click.testing import CliRunner
from mvt.ios.cli import check_sysdiagnose
@@ -54,39 +50,8 @@ def test_check_sysdiagnose_runs_explicitly_scoped_custom_module(tmp_path):
assert (output_path / "custom_sysdiagnose_module.json").exists()
def test_check_sysdiagnose_warns_without_a_custom_module(tmp_path, caplog):
# The built-in SysdiagnoseInfo alone performs no check, so the run goes
# ahead but says so.
with caplog.at_level(logging.WARNING, logger="mvt"):
result = CliRunner().invoke(
check_sysdiagnose, [str(_create_sysdiagnose_folder(tmp_path))]
)
def test_check_sysdiagnose_requires_an_explicitly_scoped_module(tmp_path):
result = CliRunner().invoke(check_sysdiagnose, [str(_create_sysdiagnose_folder(tmp_path))])
assert result.exit_code == 0
assert "No forensic sysdiagnose modules have been loaded" in caplog.text
def _create_truncated_sysdiagnose_archive(tmp_path):
folder = tmp_path / "sysdiagnose_2026.01.01_00-00-00+0000_iPhone-OS_iPhone_23A000"
folder.mkdir()
(folder / "sysdiagnose.log").write_bytes(os.urandom(200_000))
archive = tmp_path / "sysdiagnose.tar.gz"
with tarfile.open(archive, "w:gz") as tar:
tar.add(folder, arcname=folder.name)
data = archive.read_bytes()
archive.write_bytes(data[: len(data) // 2])
return archive
def test_check_sysdiagnose_reports_a_truncated_archive(tmp_path, caplog):
# A download that stopped halfway ends in EOFError from gzip, which Click
# would otherwise turn into a bare "Aborted!" with no reason given.
archive = _create_truncated_sysdiagnose_archive(tmp_path)
with caplog.at_level(logging.CRITICAL, logger="mvt"):
result = CliRunner().invoke(check_sysdiagnose, [str(archive)])
assert result.exit_code == 1
assert "Unable to read the sysdiagnose archive" in caplog.text
assert "truncated" in caplog.text
assert "Aborted!" not in result.output
assert result.exit_code != 0
assert "No custom modules support mvt-ios check-sysdiagnose" in result.output
+7 -13
View File
@@ -34,7 +34,6 @@ def _create_sysdiagnose_folder(tmp_path):
"sysdiagnose_2024.01.02_03-04-05+0200.tar.gz", encoding="utf-8"
)
(folder / "report.ips").write_text('{"bug_type": 210}\nbody', encoding="utf-8")
(folder / "._artifact.txt").write_bytes(b"\x00\x05\x16\x07AppleDouble")
return folder
@@ -46,11 +45,6 @@ def _create_sysdiagnose_archive(tmp_path, folder):
return archive_path
def _test_module(command):
(module,) = [m for m in command.executed if isinstance(m, SysdiagnoseTestModule)]
return module
def _run_command(path):
command = CmdIOSCheckSysdiagnose(
target_path=str(path), custom_modules=[SysdiagnoseTestModule]
@@ -62,30 +56,30 @@ def _run_command(path):
def test_check_sysdiagnose_from_folder(tmp_path):
command = _run_command(_create_sysdiagnose_folder(tmp_path))
assert _test_module(command).results == [
assert command.executed[0].results == [
{"content": "artifact", "timezone_offset": timedelta(hours=2).seconds}
]
assert _test_module(command).ips_files == [
assert command.executed[0].ips_files == [
{"file_path": str(tmp_path / "sysdiagnose" / "report.ips"), "bug_type": 210}
]
assert "sysdiagnose/._artifact.txt" not in command.sysdiagnose_files
def test_check_sysdiagnose_from_archive_closes_archive(tmp_path):
folder = _create_sysdiagnose_folder(tmp_path)
command = _run_command(_create_sysdiagnose_archive(tmp_path, folder))
assert _test_module(command).results == [
assert command.executed[0].results == [
{"content": "artifact", "timezone_offset": timedelta(hours=2).seconds}
]
assert _test_module(command).ips_files == [
assert command.executed[0].ips_files == [
{
"file_path": str(Path(command.extracted_sysdiagnose_path) / "report.ips"),
"file_path": str(
Path(command.extracted_sysdiagnose_path) / "report.ips"
),
"bug_type": 210,
}
]
assert command.sysdiagnose_archive is None
assert "sysdiagnose/._artifact.txt" not in command.sysdiagnose_files
def test_archive_is_extracted_once_and_unsafe_members_are_skipped(tmp_path):
Generated
+56 -4
View File
@@ -6,6 +6,22 @@ resolution-markers = [
"python_full_version < '3.15'",
]
[[package]]
name = "adb-shell"
version = "0.4.4"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "cryptography" },
{ name = "pyasn1" },
{ name = "rsa" },
]
sdist = { url = "https://files.pythonhosted.org/packages/8f/73/d246034db6f3e374dad9a35ee3f61345a6b239d4febd2a41ab69df9936fe/adb_shell-0.4.4.tar.gz", hash = "sha256:04c305f30a2ca25d5c54b3cd6ce9bb64c36e5f07967b23b3fb6aaecc851b90b6", size = 61822, upload-time = "2023-09-01T03:48:40.348Z" }
[package.optional-dependencies]
usb = [
{ name = "libusb1" },
]
[[package]]
name = "annotated-types"
version = "0.8.0"
@@ -609,14 +625,14 @@ wheels = [
[[package]]
name = "iphone-backup-decrypt"
version = "0.10.0"
version = "0.9.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "pycryptodome" },
]
sdist = { url = "https://files.pythonhosted.org/packages/01/b8/a2ec2b10d792403d36bb6661fea6357179c31ece1ed3285c442e636d6335/iphone_backup_decrypt-0.10.0.tar.gz", hash = "sha256:5f2799e58f908cce28d7929dd82231abc8acd57c25b37bce6a2d6a82e018d343", size = 20000, upload-time = "2026-09-17T21:46:44Z" }
sdist = { url = "https://files.pythonhosted.org/packages/6f/e7/bcdacdec21d628122ba240e7f742ab2175149e58672be63af55ff37a0f28/iphone_backup_decrypt-0.9.0.tar.gz", hash = "sha256:13b18fef3c8e3af627914f8c1a429bbc5555dfb0505239ba49efe99984cc0c96", size = 16125, upload-time = "2024-09-18T15:50:12.179Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/89/10/349b2a720e012921d1457e1d1c2b921b16300a5f3570c4c64d0d3787539f/iphone_backup_decrypt-0.10.0-py3-none-any.whl", hash = "sha256:1ac4345a07b104a3eacab20932b8621cd1e0164e0f03d913e66eed6ea5ee173e", size = 17400, upload-time = "2026-09-17T21:46:42Z" },
{ 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]]
@@ -761,6 +777,17 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/38/a6/800800bfed7b1fb10fc3f3d557785c3854e80d3f7a9800d784b176a1fc2d/librt-0.15.0-cp315-cp315t-win_arm64.whl", hash = "sha256:84d244b00604d17df3fc7736c327892d6bba66181254aa4087be807b6c342bdc", size = 110700, upload-time = "2026-08-07T10:49:15.499Z" },
]
[[package]]
name = "libusb1"
version = "3.4.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/89/35/f9d2a990d092d647b47540cd229e1d68432c0f51183484ca189612a4824c/libusb1-3.4.0.tar.gz", hash = "sha256:9cf5638506d54f21bf36550d97ea63189111a23c4d8078f630103a2052135f45", size = 91206, upload-time = "2026-05-16T20:59:19.315Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/d4/64/d4b59444e4d3b6979aa5eb58840634465a24b41a9ab03dcf8434c9b89551/libusb1-3.4.0-py3-none-any.whl", hash = "sha256:e83d034e44c3efe1c4599c6281d34bca50a38c12cab3b7b6217d583161a01ffd", size = 67373, upload-time = "2026-05-16T20:59:13.142Z" },
{ url = "https://files.pythonhosted.org/packages/7e/55/a838a4278fac4ee49bf670ddc59d621999a1cb2a3188fb13a23609ea4f06/libusb1-3.4.0-py3-none-win32.whl", hash = "sha256:0a1aa1416034690eb9dc9a895eda0fef44d853bca1053eb1de50a5906684846d", size = 129704, upload-time = "2026-05-16T20:59:15.243Z" },
{ url = "https://files.pythonhosted.org/packages/5f/08/02aecf6dad627534a5835244ece14d7f187ef430354d9b8551199934d059/libusb1-3.4.0-py3-none-win_amd64.whl", hash = "sha256:b7dcc1f324a895af6aac708bc5513a17373f97349cc2ab8a277519c788bd18ca", size = 141212, upload-time = "2026-05-16T20:59:17.286Z" },
]
[[package]]
name = "markdown"
version = "3.10.3"
@@ -989,11 +1016,13 @@ wheels = [
name = "mvt"
source = { editable = "." }
dependencies = [
{ name = "adb-shell", extra = ["usb"] },
{ name = "appdirs" },
{ name = "betterproto2" },
{ name = "click" },
{ name = "cryptography" },
{ name = "iphone-backup-decrypt" },
{ name = "libusb1" },
{ name = "nskeyedunarchiver" },
{ name = "packaging" },
{ name = "pyahocorasick" },
@@ -1034,11 +1063,13 @@ docs = [
[package.metadata]
requires-dist = [
{ name = "adb-shell", extras = ["usb"], specifier = "==0.4.4" },
{ name = "appdirs", specifier = "==1.4.4" },
{ name = "betterproto2", specifier = "==0.10.0" },
{ name = "click", specifier = "==8.4.2" },
{ name = "cryptography", specifier = "==50.0.0" },
{ name = "iphone-backup-decrypt", specifier = "==0.10.0" },
{ name = "iphone-backup-decrypt", specifier = "==0.9.0" },
{ name = "libusb1", specifier = "==3.4.0" },
{ name = "nskeyedunarchiver", specifier = "==1.5.2" },
{ name = "packaging", specifier = "==26.3" },
{ name = "pyahocorasick", specifier = "==2.3.1" },
@@ -1243,6 +1274,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/c5/96/37c50ac951bb0260ec38d8d12e5b51587ef1ef4035c279088f2771544b28/pyahocorasick-2.3.1-cp314-cp314-win_amd64.whl", hash = "sha256:4acb11a0a2ff10519465749d22ad70789e9fe7f81dc8fe9957a8868e499e18ab", size = 35987, upload-time = "2026-04-27T16:32:07.08Z" },
]
[[package]]
name = "pyasn1"
version = "0.6.4"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/a4/9a/23310166d960def5897e91fe20e5b724601b02a22e84ba1f94232c0b7f67/pyasn1-0.6.4.tar.gz", hash = "sha256:9c447d8431c947fe4c8febc4ed9e760bc29011a5b01e5c74b67025bd9fb8ce81", size = 151262, upload-time = "2026-07-09T01:12:33.988Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/9a/3b/6163796d69c3977d1e4287bea4a6979161cbbdd170ebb430511e8e1999ce/pyasn1-0.6.4-py3-none-any.whl", hash = "sha256:deda9277cfd454080ec40b207fb6df82206a3a2688735233cdcd8d3d565f088b", size = 84410, upload-time = "2026-07-09T01:12:32.92Z" },
]
[[package]]
name = "pycparser"
version = "3.0"
@@ -1644,6 +1684,18 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/82/3b/64d4899d73f91ba49a8c18a8ff3f0ea8f1c1d75481760df8c68ef5235bf5/rich-15.0.0-py3-none-any.whl", hash = "sha256:33bd4ef74232fb73fe9279a257718407f169c09b78a87ad3d296f548e27de0bb", size = 310654, upload-time = "2026-04-12T08:24:02.83Z" },
]
[[package]]
name = "rsa"
version = "4.9.1"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "pyasn1" },
]
sdist = { url = "https://files.pythonhosted.org/packages/da/8a/22b7beea3ee0d44b1916c0c1cb0ee3af23b700b6da9f04991899d0c555d4/rsa-4.9.1.tar.gz", hash = "sha256:e7bdbfdb5497da4c07dfd35530e1a902659db6ff241e39d9953cad06ebd0ae75", size = 29034, upload-time = "2025-04-16T09:51:18.218Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/64/8d/0133e4eb4beed9e425d9a98ed6e081a55d195481b7632472be1af08d2f6b/rsa-4.9.1-py3-none-any.whl", hash = "sha256:68635866661c6836b8d39430f97a996acbd61bfa49406748ea243539fe239762", size = 34696, upload-time = "2025-04-16T09:51:17.142Z" },
]
[[package]]
name = "ruff"
version = "0.16.2"