Compare commits

..
Author SHA1 Message Date
Donncha Ó Cearbhaill 06cea62a40 Record wall time and peak memory growth per module
Command.run() times every module and measures how much the process's
peak resident set size grew while it ran. Both are logged at INFO in
one line per module and written to info.json under module_stats, in
run order, so any run says which module cost what and CI can compare a
branch against main.

ru_maxrss is the process peak, so a module's growth is how much it
raised the high-water mark, which is what matters on a host without
swap; a module running after a bigger one reports 0. Where the resource
module is unavailable the growth is reported as 0.
2026-09-16 00:45:47 +02:00
Donncha Ó Cearbhaill eceafdc785 Keep the test suite out of the user's MVT folders (#923)
The tests read the settings file of whoever runs them, and importing
mvt.common.config writes it back. On a machine whose config.yaml sets
NETWORK_ACCESS_ALLOWED to false, the plugin update and URL batch tests
fail before their mocked requests are reached, and every Command run in
the suite parses the indicators downloaded on that machine, which made
the suite take minutes instead of seconds.

MVT_CONFIG_FOLDER and MVT_DATA_FOLDER in the environment now relocate
the settings file and the downloaded indicators with their update-check
state. The test conftest points both at a throwaway folder before any
mvt module is imported, and removes it at exit. Subprocesses started by
the tests inherit the variables; the isolated interpreter helper already
gives them a temporary home.

On the machine that prompted this the suite goes from 6 failures in six
minutes to none in seven seconds, and config.yaml is left alone.
2026-09-08 01:04:43 +01:00
Donncha Ó Cearbhaill ffae240355 Skip AppleDouble sidecars when listing a sysdiagnose (#922)
Device-generated sysdiagnose archives carry a ._name entry beside every
file that has extended attributes, an ACL or Finder info; one iOS 26
archive held 1234 of them among 3648 members, and the count grows with
each release. bsdtar folds them back into the file on extraction and
hides them from listings, but tarfile returns them as regular members,
so check-sysdiagnose extracted them and handed them to every module.
A module that globs for plists or logs then tries to parse AppleDouble
headers and logs one warning per sidecar.

Leave them out of the file list, both for archives and for folders
extracted on a system that keeps them as files.
2026-09-08 01:03:05 +01:00
3623e430f4 Defer indicator loading until first use (#919)
* Defer indicator loading until first use

* Load indicators once at the start of a run

The lazy property loads indicators wherever they are first read, which
in Command.run() is inside the module loop, after init(). For
check-androidqf that means the whole acquisition is walked before a
missing --iocs file is reported, and the STIX parsing lines land between
the module list and the first module.

Read the indicators once after the module list is settled, so that a bad
--module name still loads nothing, and hand the same object to every
module. check-iocs reads them once too, so that a wrong --iocs path is
reported even when no stored result matches a module.

* Drop two test assertions the change does not need

Retrying after a failed indicator load is incidental to the property
rather than a requirement, so nothing should pin it. The exact wording
of the check-backup rejection belongs to that command's own tests; exit
code 1 already shows the path was rejected before indicators were
loaded.

* Create the output folder and command.log when a run starts

Command.__init__ created the --output folder and attached the
command.log handler, so `--list-modules -o out` or a rejected target
path left an empty folder behind for a run that never happened.

Do both at the top of run(), before the module list is resolved so that
its warnings still reach the log. The nested commands run by
check-androidqf re-attach the handler at the start of their runs, as
they did at construction. Lines the CLI logs between construction and
run(), such as the target path being checked, no longer reach
command.log; info.json records the target path.

This is the remaining part of #888.

* Cache the indicators with functools.cached_property

A property with a setter and a backing attribute does by hand what
cached_property does: compute on first read, store the result on the
instance, and accept assignment, which is how nested commands receive
their parent's indicators. A failed load is still not cached.

* Announce the target from the command so that it reaches command.log

The CLI logged which backup, filesystem or acquisition it was about to
check just before run(), which is now before command.log exists, so the
line only reached the console. Each command logs it from init() instead,
which runs once the log is attached, and run() logs the module list
after init() so that the target still comes first. Nested commands
without a target path log nothing, as before.

* Log the Android backup path from the typed local

mypy cannot determine the type of target_path in this command, which the
surrounding lines already silence, so read the announced path from the
local that carries the type instead of adding another ignore.

---------

Co-authored-by: bitmeta69 <206962233+bitmeta69@users.noreply.github.com>
Co-authored-by: besendorf <janik@besendorf.org>
Co-authored-by: Donncha Ó Cearbhaill <donncha.ocearbhaill@amnesty.org>
2026-09-08 01:01:02 +01:00
Donncha Ó Cearbhaill 60e652bafc Read bugreport file timestamps in the device's timezone (#921)
A bugreport zip stores each entry's time as the device's wall clock, and
an unpacked bugreport's mtimes are whatever the extraction left. The
bugreport modules read both as naive local datetimes, so a tombstone's
file_timestamp moved with the analysing machine's timezone and sat three
hours from the crash time the tombstone itself records for a device in
Nairobi.

check-bugreport now resolves the device's timezone once, from --timezone
or from persist.sys.timezone in the dumpstate's SYSTEM PROPERTIES, and
hands it to the modules as device_timezone the way check-androidqf does;
a zone already known, as when androidqf drives the bugreport inside its
own archive, is kept. Zip entry times are read in that zone and the
mtimes of an unpacked bugreport as UTC instants, so convert_datetime_to_iso
writes both in UTC. Without a zone the wall clock is kept naive and a
warning says so, and an unpacked bugreport gets a warning that its file
timestamps are the extraction's, not the device's. BugReportTimestamps
uses the same reading instead of parsing the properties itself.
2026-09-08 00:08:30 +01:00
Donncha Ó Cearbhaill 92f780df5a Report an unreadable sysdiagnose archive instead of aborting silently (#920)
A sysdiagnose tarball whose download stopped halfway ends in an EOFError
from gzip while check-sysdiagnose extracts it. Click turns EOFError into
click.Abort, so the command printed nothing but "Aborted!", even with
-v. The extraction now catches the read errors an archive can raise,
names the file and the reason at critical level, says the file may be
truncated or not a gzip tarball, and exits 1. A plain .tar given to the
gzip reader gets the same message rather than a traceback.
2026-09-07 23:58:43 +01:00
Donncha Ó Cearbhaill 2eb40b85cf Add a SysdiagnoseInfo module to check-sysdiagnose (#917)
* Add a SysdiagnoseInfo module to check-sysdiagnose

check-sysdiagnose had no module of its own: it prepared the archive for
plugin modules and refused to run without one. SysdiagnoseInfo is the
first built-in module. It writes sysdiagnose_info.json with details
about the device and the archive: product type and model, iOS version
and build, serial number, IMEI, MEID and UDID from remotectl_dumpstate.txt
and the mobile activation request, the Apple account name and email from
the App Store daemon database, and the archive's original file name and
creation time from sysdiagnose.log. The build is checked against the
known iOS versions the way BackupInfo does.

The App Store database is copied out of the archive together with its
-wal and -shm sidecars before it is opened, so rows still in the
write-ahead log are read.

With a built-in module the command's list is never empty, so the "no
custom modules" error and its test go. The module joins
IOS_CHECK_IOCS_MODULES like every other module that writes a results
file.

* Note that newer sysdiagnoses lack the App Store daemon database

* Keep refusing check-sysdiagnose runs without a custom module

* Warn instead of refusing when no forensic sysdiagnose module is loaded
2026-09-05 23:45:14 +02:00
Donncha Ó Cearbhaill b9055e365c Merge pull request #916 from mvt-project/fix/settings-env-persist
Stop persisting environment variables to config.yaml
2026-09-05 16:44:40 +02:00
Donncha Ó Cearbhaill fc217e9608 Merge branch 'main' into fix/settings-env-persist 2026-09-05 16:43:43 +02:00
Donncha Ó Cearbhaill d7148c03b4 Stop persisting environment variables to config.yaml
MVTSettings.initialise() constructs the settings once with load_env=False
and writes the result to config.yaml, so that values taken from MVT_*
environment variables are never persisted, then constructs them again
with the environment applied.

Since #716 settings_customise_sources() has added env_settings
unconditionally, making load_env dead code. Any MVT_* variable set for a
single run, including MVT_IOS_BACKUP_PASSWORD, MVT_ANDROID_BACKUP_PASSWORD
and MVT_VT_API_KEY, was written in plaintext to config.yaml and read back
on every later run.

Only add env_settings when load_env is true, and add a regression test.

Fixes #915
2026-09-05 16:41:50 +02:00
Donncha Ó Cearbhaill 0ddcab904d Merge pull request #914 from DonnchaC/feature/bugreport-settings-records
Parse dumpsys settings as per-record results
2026-09-05 16:32:02 +02:00
Donncha Ó Cearbhaill 858496e60b End a settings record at a blank line
dumpsys prints a blank line after every namespace block and after a
change history, and the generation registry and any vendor dumps follow
the last block before the section trailer. A record ran until the next
`_id:` line, heading or trailer, so the last row of the section
absorbed those dumps into whichever field came last.

A blank line now closes the record being read; lines that follow it
without an `_id:` are skipped until the next row or heading. The fixture
carries a generation registry after its last block, modelled on the
AOSP dump, and the last row is pinned to exactly its own fields.
2026-09-05 16:17:43 +02:00
Donncha Ó Cearbhaill e5fced2a08 Peel tag and restore flags off settings rows
A dumpsys settings row can end in `tag:` and, on some vendor builds, in
`isValuePreservedInRestore:` or a bare `notPreservedInRestore` token.
The parser only peeled `default:` and `defaultSystemSet:` off the end
of a record, so on a row without a default those tokens stayed inside
the value, and on a row with one they landed in `defaultSystemSet`. A
value of `0 tag:null` is not the safe value `0`, so a setting at its
safe value was reported as dangerous; these are the false positives
measured in #912.

The metadata keys are ordered fields like the rest of the record, so
`_split_fields` reads them now, which also replaces the separate
handling of the default. The bare token has no `key:` shape and is
printed last, so it is stripped first and recorded as
`isValuePreservedInRestore: false`.

The bugreport fixture gains the three row shapes from #912: a restore
flag after `defaultSystemSet:`, and a `tag:` or a bare token directly
after the value. Two of them sit on dangerous settings at their safe
value, so the unchanged alert count of one is the false-positive check.
2026-09-05 15:56:55 +02:00
Donncha Ó Cearbhaill fe505c63a6 Merge branch 'main' into feature/bugreport-settings-records 2026-09-05 00:59:11 +02:00
Donncha Ó Cearbhaill 25a571c462 Merge pull request #913 from mvt-project/fix/coverage-logging
Fix CI coverage reports and fix matrix testing each Python version
2026-09-04 20:34:18 +02:00
Donncha Ó Cearbhaill 03012d2c2d Fix CI coverage comment and test each matrix Python version
The coverage comment never posted: the per-file table with a link on
every file and missing line range exceeds GitHub's 65536-character
comment limit, so PRs got only a badge. Every matrix job also raced to
post the same comment, and the action was unpinned at @main.

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

Set UV_PYTHON from the matrix. Without it, .python-version pins 3.10
and `uv run` rebuilt the venv with 3.10 after `uv sync --python X`, so
all five matrix jobs were testing on Python 3.10.
2026-09-04 20:25:07 +02:00
Donncha Ó Cearbhaill 0e231eefaf Parse dumpsys settings as per-record results
The `dumpsys settings` parser matched a single regex per line, which
truncated every value that spans more than one line and, when
`defaultSystemSet:` did not fall on the first line, left the trailing
`default:` metadata inside the value. It also keyed results by setting
name within a namespace, so a name recorded twice kept only the last row
and a row without a `pkg:` field was dropped entirely.

Replace it with a line loop that accumulates one record at a time and
splits the `key:value` fields once the whole record has been read.
Results become a list of records carrying the fields dumpsys prints:
namespace, user, _id, name, value, pkg, default and defaultSystemSet,
plus the per-setting change history. History timestamps are printed
without a year, so they are resolved against the "ending at:" time of
the section and serialized into the timeline. This shows which package
changed a security-relevant setting, and when.

The androidqf settings module shares this artifact, so it now emits the
same record shape.
2026-09-04 17:44:51 +02:00
besendorf a463e8509c Merge pull request #774 from mvt-project/backup_extraction
Replace iOSbackup with iphone_backup_decrypt
2026-09-02 19:29:12 +02:00
38 changed files with 1468 additions and 204 deletions

No files matched your search

+19 -4
View File
@@ -5,6 +5,10 @@ on:
pull_request:
branches: [ main ]
permissions:
contents: read
pull-requests: write # coverage comment
jobs:
build:
name: Run Python Tests
@@ -13,6 +17,10 @@ jobs:
fail-fast: false
matrix:
python-version: ['3.10', '3.11', '3.12', '3.13', '3.14']
env:
# Takes precedence over .python-version, which otherwise makes `uv run`
# rebuild the venv with 3.10 and test every matrix entry on 3.10.
UV_PYTHON: ${{ matrix.python-version }}
steps:
- uses: actions/checkout@v7
@@ -26,16 +34,23 @@ jobs:
python-version: ${{ matrix.python-version }}
- name: Install Python dependencies
run: |
uv sync --locked --group dev --python ${{ matrix.python-version }}
uv sync --locked --group dev
- name: Test with pytest
run: |
set -o pipefail
make test-ci | tee pytest-coverage.txt
- name: Coverage job summary
run: uv run coverage report --format=markdown --show-missing --skip-covered >> "$GITHUB_STEP_SUMMARY"
- name: Pytest coverage comment
continue-on-error: true # Workflows running on a fork can't post comments
uses: MishaKav/pytest-coverage-comment@main
if: github.event_name == 'pull_request'
# One comment per PR, not one per matrix entry. PRs from forks get a
# read-only token and can't post; the job summary above still works.
if: github.event_name == 'pull_request' && matrix.python-version == '3.13'
continue-on-error: true
uses: MishaKav/pytest-coverage-comment@v1.12.2
with:
pytest-coverage-path: ./pytest-coverage.txt
junitxml-path: ./pytest.xml
# The full table with per-line links exceeds GitHub's 65536-char comment limit.
report-only-changed-files: true
remove-links-to-lines: true
+8
View File
@@ -435,3 +435,11 @@ 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*.
+7 -4
View File
@@ -1,10 +1,13 @@
# Check an iOS Sysdiagnose
`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
`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
[plugin packages](../development/index.md#installed-module-packages) which
declare support for it. Install at least one such package first.
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.
The command accepts either an extracted sysdiagnose directory or the original
gzip-compressed tar archive.
+236 -36
View File
@@ -4,6 +4,11 @@
# https://license.mvt.re/1.1/
import re
from datetime import datetime
from typing import Optional, Sequence
from mvt.common.module_types import ModuleAtomicResult, ModuleSerializedResult
from mvt.common.utils import convert_datetime_to_iso
from .artifact import AndroidArtifact
@@ -60,45 +65,240 @@ ANDROID_DANGEROUS_SETTINGS = [
},
]
# dumpsys prints the fields of a setting record, and of a change history entry,
# always in this order and separated by a single space. After the value come
# `default:` and `defaultSystemSet:` when a default is recorded, then `tag:`;
# some vendor builds add whether the value survives a restore, either as
# `isValuePreservedInRestore:` or as a bare `notPreservedInRestore` token.
SETTING_FIELDS = (
"_id",
"name",
"pkg",
"value",
"default",
"defaultSystemSet",
"tag",
"isValuePreservedInRestore",
)
HISTORY_FIELDS = ("time", "mode", "oldValue", "newValue", "package")
NAMESPACE_PATTERN = re.compile(
r"^(CONFIG|GLOBAL|SECURE|SYSTEM) SETTINGS \(user (\d+)\)$"
)
SECTION_END_PATTERN = re.compile(r"ending at: (\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})")
class Settings(AndroidArtifact):
def parse(self, content: str) -> None:
self.results: dict[str, dict[str, str]] = {}
namespace: str | None = None
for line in content.splitlines():
heading = re.match(
r"^(CONFIG|GLOBAL|SECURE|SYSTEM) SETTINGS \(user (\d+)\)$",
line.strip(),
)
if heading:
namespace = f"{heading.group(1).lower()}:user_{heading.group(2)}"
self.results[namespace] = {}
"""Parser for the `dumpsys settings` output.
Every row of the settings provider becomes one result, keeping the fields
dumpsys prints alongside the value: the row id, the package which recorded
the setting, the default, the tag, and the change history. A setting name
can appear more than once within a namespace, so results are a list rather
than a mapping.
"""
def serialize(self, result: ModuleAtomicResult) -> ModuleSerializedResult:
records = []
for entry in result.get("history", []):
if not entry.get("timestamp"):
continue
if namespace is None or not line.startswith("_id:"):
continue
setting = re.match(
r"^_id:\S+\s+name:(.*?)\s+pkg:.*?\s+value:(.*?)"
r"(?:\s+default:.*\s+defaultSystemSet:(?:true|false))?$",
line,
records.append(
{
"timestamp": entry["timestamp"],
"module": self.__class__.__name__,
"event": "settings_change",
"data": (
f"{result.get('namespace')} setting "
f'"{result.get("name")}" changed from '
f'"{entry.get("oldValue")}" to "{entry.get("newValue")}" '
f"by {entry.get('pkg')}"
),
}
)
if setting:
self.results[namespace][setting.group(1)] = setting.group(2)
return records
def check_indicators(self) -> None:
for namespace, settings in self.results.items():
for key, value in settings.items():
for danger in ANDROID_DANGEROUS_SETTINGS:
# Check if one of the dangerous settings is using an unsafe
# value (different than the one specified).
if danger["key"] == key and danger["safe_value"] != value:
self.alertstore.medium(
f'Found suspicious "{namespace}" setting "{key} = {value}" ({danger["description"]})',
"",
{
"namespace": namespace,
"key": key,
"value": value,
"description": danger["description"],
},
)
break
for result in self.results:
name = result.get("name")
value = result.get("value")
for danger in ANDROID_DANGEROUS_SETTINGS:
# Check if one of the dangerous settings is using an unsafe
# value (different than the one specified).
if danger["key"] != name or danger["safe_value"] == value:
continue
history = result.get("history") or []
self.alertstore.medium(
f'Found suspicious "{result.get("namespace")}" setting '
f'"{name} = {value}" ({danger["description"]})',
history[-1]["timestamp"] if history else "",
result,
)
break
def parse(self, content: str) -> None:
self.results: list[ModuleAtomicResult] = []
section_end = self._parse_section_end(content)
namespace: Optional[str] = None
user: Optional[str] = None
record_lines: list[str] = []
history_lines: list[str] = []
in_history = False
def flush() -> None:
nonlocal record_lines, history_lines, in_history
if record_lines:
self.results.append(
self._build_record(
namespace, user, record_lines, history_lines, section_end
)
)
record_lines = []
history_lines = []
in_history = False
for line in content.splitlines():
heading = NAMESPACE_PATTERN.match(line.strip())
if heading:
flush()
namespace = heading.group(1).lower()
user = heading.group(2)
continue
if line.startswith("--------- "):
# dumpsys closes every section with a duration trailer.
flush()
namespace = None
continue
if namespace is None:
continue
if not line.strip():
# dumpsys prints a blank line after every namespace block and
# after a change history, and other dumps such as the
# generation registry follow the last block, so a blank line
# closes the record being read.
flush()
continue
if line.startswith("_id:"):
flush()
record_lines = [line]
continue
if not record_lines:
continue
stripped = line.strip()
if stripped.startswith("History ("):
in_history = True
continue
if in_history:
if stripped.startswith("time:"):
history_lines.append(stripped)
elif stripped and history_lines:
# A history entry can be wrapped over several lines.
history_lines[-1] += " " + stripped
continue
# Anything else continues the value of the record being read.
record_lines.append(line)
flush()
@staticmethod
def _split_fields(text: str, keys: Sequence[str]) -> dict[str, str]:
"""Split the `key:value` fields of one record.
Values are free-form and may contain spaces and newlines, so a field
runs up to the start of the next key which is actually present. Keys
dumpsys did not print are skipped.
"""
fields: dict[str, str] = {}
key = keys[0]
if not text.startswith(f"{key}:"):
return fields
remainder = text[len(key) + 1 :]
for next_key in keys[1:]:
value, separator, rest = remainder.partition(f" {next_key}:")
if separator:
fields[key] = value
key, remainder = next_key, rest
fields[key] = remainder
return fields
@staticmethod
def _parse_section_end(content: str) -> Optional[datetime]:
"""Return the time the settings section was dumped, if reported."""
match = SECTION_END_PATTERN.search(content)
if not match:
return None
try:
return datetime.strptime(match.group(1), "%Y-%m-%d %H:%M:%S")
except ValueError:
return None
@staticmethod
def _resolve_timestamp(
value: str, section_end: Optional[datetime]
) -> Optional[str]:
"""Add the missing year to a `MM-DD HH:MM:SS.mmm` history timestamp.
dumpsys prints the change history without a year, so it is resolved
against the time the section was dumped: the most recent matching date
at or before that time.
"""
if section_end is None:
return None
try:
partial = datetime.strptime(value, "%m-%d %H:%M:%S.%f")
timestamp = partial.replace(year=section_end.year)
if timestamp > section_end:
timestamp = partial.replace(year=section_end.year - 1)
except ValueError:
return None
return convert_datetime_to_iso(timestamp)
def _parse_history(
self, line: str, section_end: Optional[datetime]
) -> ModuleAtomicResult:
fields = self._split_fields(line, HISTORY_FIELDS)
return {
"timestamp": self._resolve_timestamp(fields.get("time", ""), section_end),
"oldValue": fields.get("oldValue"),
"newValue": fields.get("newValue"),
"pkg": fields.get("package"),
}
def _build_record(
self,
namespace: Optional[str],
user: Optional[str],
record_lines: list[str],
history_lines: list[str],
section_end: Optional[datetime],
) -> ModuleAtomicResult:
text = "\n".join(record_lines).rstrip()
# The bare `notPreservedInRestore` token has no `key:` shape and is
# printed last, so peel it off before splitting the fields.
head = text.removesuffix(" notPreservedInRestore")
record: ModuleAtomicResult = {"namespace": namespace, "user": user}
record.update(self._split_fields(head, SETTING_FIELDS))
if head != text:
record["isValuePreservedInRestore"] = "false"
record["history"] = [
self._parse_history(entry, section_end) for entry in history_lines
]
return record
+17 -8
View File
@@ -150,6 +150,16 @@ 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
@@ -160,6 +170,7 @@ def check_bugreport(
list_modules,
module,
load_module,
timezone,
verbose,
bugreport_path,
):
@@ -167,12 +178,18 @@ 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],
@@ -183,8 +200,6 @@ 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:
@@ -259,8 +274,6 @@ 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()
@@ -342,8 +355,6 @@ 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()
@@ -424,8 +435,6 @@ 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,6 +82,8 @@ 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,6 +129,7 @@ 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"
+50 -5
View File
@@ -9,6 +9,7 @@ from pathlib import Path
from typing import List, Optional
from zipfile import ZipFile
from mvt.android.artifacts.getprop import GetProp
from mvt.android.modules.bugreport.base import BugReportModule
from mvt.common.command import Command
from mvt.common.indicators import Indicators
@@ -88,13 +89,57 @@ class CmdAndroidCheckBugreport(Command):
self.__files.append(file_name)
def init(self) -> None:
if not self.target_path:
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"])
return
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)
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."
)
def module_init(self, module: BugReportModule) -> None: # type: ignore[override]
if self.__format == "zip":
@@ -63,6 +63,8 @@ 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")
@@ -3,11 +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 logging
from typing import Optional
from mvt.android.artifacts.settings import Settings as SettingsArtifact
from mvt.common.module_types import ModuleResults
from .base import AndroidQFModule
@@ -15,43 +11,23 @@ from .base import AndroidQFModule
class AQFSettings(SettingsArtifact, AndroidQFModule):
"""This module analyse setting files"""
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 run(self) -> None:
for setting_file in self._get_files_by_pattern("*/settings_*.txt"):
namespace = setting_file[setting_file.rfind("_") + 1 : -4]
self.results[namespace] = {}
data = self._get_file_content(setting_file)
for line in data.decode("utf-8").splitlines():
line = line.strip()
try:
key, value = line.split("=", 1)
except ValueError:
name, separator, value = line.strip().partition("=")
if not separator:
continue
try:
self.results[namespace][key] = value
except IndexError:
continue
self.results.append(
{
"namespace": namespace,
"user": None,
"name": name,
"value": value,
}
)
self.log.info(
"Identified %d settings", sum([len(val) for val in self.results.values()])
)
self.log.info("Identified %d settings", len(self.results))
+26 -6
View File
@@ -9,6 +9,7 @@ import os
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
@@ -125,12 +126,31 @@ class BugReportModule(MVTModule):
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)
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)
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
)
@@ -4,15 +4,12 @@
# 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):
@@ -41,41 +38,28 @@ class BugReportTimestamps(FileTimestampsArtifact, BugReportModule):
def run(self) -> None:
filesystem_files = self._get_files_by_pattern("FS/*")
timezone_name = None
dumpstate = self._get_dumpstate_file()
if dumpstate:
section = self.extract_command_section(
dumpstate.decode("utf-8", errors="replace"),
"------ SYSTEM PROPERTIES",
)
properties = GetProp()
properties.parse(section)
timezone_name = properties.get_device_timezone()
timezone = None
if timezone_name:
try:
timezone = ZoneInfo(timezone_name)
except ZoneInfoNotFoundError:
self.log.warning("Unknown device timezone %s", timezone_name)
self.results = []
for file in filesystem_files:
# Only the modification time is available in the zip file metadata.
# The timezone is the local timezone of the machine the phone.
# 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.
modification_time = self._get_file_modification_time(file)
utc_time = None
if timezone is not None:
utc_time = convert_datetime_to_iso(
modification_time.replace(tzinfo=timezone).astimezone(
datetime.timezone.utc
)
)
self.results.append(
{
"path": file,
"modified_time": convert_datetime_to_iso(modification_time),
"modified_time_utc": utc_time,
"timezone": timezone_name,
"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"
),
"timestamp_source": (
"zip_metadata" if self.zip_archive else "filesystem_metadata"
),
@@ -18,5 +18,4 @@ class Settings(SettingsArtifact, BugReportModule):
data.decode("utf-8", errors="replace"), "DUMP OF SERVICE settings:"
)
self.parse(section)
count = sum(len(settings) for settings in self.results.values())
self.log.info("Identified %d Android settings", count)
self.log.info("Identified %d Android settings", len(self.results))
+6 -2
View File
@@ -56,6 +56,10 @@ 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
@@ -83,8 +87,8 @@ class CmdCheckIOCS(Command):
log.warning("No result from this module, skipping it")
continue
if self.iocs.total_ioc_count > 0:
m.indicators = self.iocs
if iocs.total_ioc_count > 0:
m.indicators = iocs
m.indicators.log = m.log
try:
+55 -10
View File
@@ -7,7 +7,9 @@ import json
import logging
import os
import sys
import time
from datetime import datetime
from functools import cached_property
from heapq import heappop, heappush
from typing import Any, Optional
@@ -35,6 +37,17 @@ from .utils import (
from .version import MVT_VERSION
def _peak_rss_mb() -> float:
"""The process's peak resident set size so far, in MB (0 where unknown)."""
try:
import resource
except ImportError:
return 0.0
peak = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
# ru_maxrss is in bytes on macOS and in kilobytes elsewhere.
return peak / (1024 * 1024 if sys.platform == "darwin" else 1024)
class Command:
def __init__(
self,
@@ -76,6 +89,9 @@ class Command:
# declaring `replaces` to the module which took their place.
self.module_replacements: dict[type[MVTModule], type[MVTModule]] = {}
# Wall time and peak-RSS growth of every module run, in run order.
self.module_stats: list[dict[str, Any]] = []
# This list will contain all executed modules.
# We can use this to reference e.g. self.executed[0].results.
self.executed: list[MVTModule] = []
@@ -84,18 +100,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:
@@ -189,6 +205,7 @@ class Command:
"date": convert_datetime_to_iso(datetime.now()),
"ioc_files": [],
"hashes": [],
"module_stats": self.module_stats,
}
for coll in self.iocs.ioc_collections:
@@ -710,17 +727,30 @@ 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
self._log_loaded_modules(ordered_modules)
# 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
# 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:
@@ -740,8 +770,8 @@ class Command:
for dependency, resolved in self._module_dependencies(module)
}
if self.iocs.total_ioc_count:
m.indicators = self.iocs
if iocs.total_ioc_count:
m.indicators = iocs
m.indicators.log = m.log
if self.serial:
@@ -752,6 +782,8 @@ class Command:
except NotImplementedError:
pass
started = time.perf_counter()
peak_before = _peak_rss_mb()
try:
run_module(m)
except EncryptedBackupError:
@@ -761,6 +793,19 @@ class Command:
)
return
stats = {
"module": module.__name__,
"seconds": round(time.perf_counter() - started, 2),
"peak_rss_growth_mb": round(_peak_rss_mb() - peak_before),
}
self.module_stats.append(stats)
self.log.info(
"Module %s ran in %.1fs and raised the peak memory by %d MB",
stats["module"],
stats["seconds"],
stats["peak_rss_growth_mb"],
)
self.executed.append(m)
executed_by_type[module] = m
self.timeline.extend(m.timeline)
+11 -6
View File
@@ -12,7 +12,9 @@ from pydantic_settings import (
YamlConfigSettingsSource,
)
MVT_CONFIG_FOLDER = user_config_dir("mvt")
# 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_PATH = os.path.join(MVT_CONFIG_FOLDER, "config.yaml")
@@ -59,13 +61,16 @@ class MVTSettings(BaseSettings):
dotenv_settings: PydanticBaseSettingsSource,
file_secret_settings: PydanticBaseSettingsSource,
) -> Tuple[PydanticBaseSettingsSource, ...]:
yaml_source = YamlConfigSettingsSource(settings_cls, MVT_CONFIG_PATH)
sources: Tuple[PydanticBaseSettingsSource, ...] = (
yaml_source,
YamlConfigSettingsSource(settings_cls, MVT_CONFIG_PATH),
init_settings,
)
# Always load env variables by default
sources = (env_settings,) + sources
# Load env variables only when asked to. initialise() constructs the
# settings once without them so that what gets written back to
# config.yaml never includes values taken from the environment.
# init_settings() returns the keyword arguments passed to the constructor.
if init_settings().get("load_env", True):
sources = (env_settings,) + sources
return sources
def save_settings(
@@ -92,7 +97,7 @@ class MVTSettings(BaseSettings):
Afterwards we load the settings again, this time including the env variables.
"""
# Set invalid env prefix to avoid loading env variables.
# Construct the settings without env variables so they are not persisted.
settings = cls(load_env=False)
settings.save_settings()
+7 -5
View File
@@ -18,7 +18,9 @@ from appdirs import user_data_dir
from .config import settings
from .url import URL
MVT_DATA_FOLDER = user_data_dir("mvt")
# 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_INDICATORS_FOLDER = os.path.join(MVT_DATA_FOLDER, "indicators")
logger = logging.getLogger(__name__)
@@ -71,7 +73,9 @@ 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(
@@ -518,9 +522,7 @@ 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)
+8 -10
View File
@@ -318,8 +318,6 @@ 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()
@@ -386,8 +384,6 @@ 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()
@@ -449,18 +445,20 @@ def check_sysdiagnose(
custom_modules=custom_modules,
)
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\"),)."
# 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 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,6 +59,9 @@ 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,5 +51,8 @@ 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
+27 -2
View File
@@ -7,7 +7,9 @@ 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
@@ -16,6 +18,8 @@ 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__)
@@ -52,6 +56,7 @@ 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] = []
@@ -76,11 +81,15 @@ 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)
@@ -94,8 +103,19 @@ class CmdIOSCheckSysdiagnose(Command):
self.log.info("Parsing sysdiagnose archive. This might take a while...")
self.sysdiagnose_format = "tar"
self.sysdiagnose_archive = tarfile.open(self.target_path, "r:gz")
self._extract_sysdiagnose_archive()
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)
def _extract_sysdiagnose_archive(self) -> None:
archive = self.sysdiagnose_archive
@@ -119,6 +139,11 @@ 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():
+2 -1
View File
@@ -16,7 +16,8 @@ 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
BACKUP_MODULES + FS_MODULES + MIXED_MODULES + SYSDIAGNOSE_MODULES
)
@@ -3,4 +3,9 @@
# 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]
@@ -0,0 +1,228 @@
# 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)
+170
View File
@@ -0,0 +1,170 @@
# Mobile Verification Toolkit (MVT)
# Copyright (c) 2021-2023 The MVT Authors.
# Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/
from mvt.android.artifacts.settings import Settings
from ..utils import get_artifact
def parse_bugreport_settings() -> Settings:
settings = Settings()
with open(get_artifact("android_data/bugreport/dumpstate.txt")) as handle:
data = handle.read()
settings.parse(settings.extract_dumpsys_section(data, "DUMP OF SERVICE settings:"))
return settings
def find(settings: Settings, name: str) -> list:
return [result for result in settings.results if result["name"] == name]
class TestSettingsArtifact:
def test_parsing(self):
settings = parse_bugreport_settings()
assert len(settings.results) == 12
assert {result["namespace"] for result in settings.results} == {
"config",
"global",
"secure",
}
assert settings.results[0] == {
"namespace": "config",
"user": "0",
"_id": "682",
"name": "namespace_one/blocked_components",
"pkg": "com.example.services",
"value": (
"com.android.settings,com.android.vending,\n"
"com.example.dialer,\n"
"com.example.camera"
),
"default": (
"com.android.settings,\n"
" com.android.vending,\n"
" com.example.dialer"
),
"defaultSystemSet": "false",
"history": [],
}
def test_multiline_values_are_kept_whole(self):
settings = parse_bugreport_settings()
assert find(settings, "namespace_one/allowed_packages")[0]["value"] == (
"com.example.messaging,\ncom.example.chat"
)
assert find(settings, "widget_instance_data")[0]["value"] == (
'{\n "version": 1,\n "data": [\n {\n "number": 10000,\n'
' "package_name": "com.example.widget"\n }\n ]\n}'
)
def test_trailing_default_is_not_part_of_the_value(self):
settings = parse_bugreport_settings()
record = find(settings, "namespace_one/streaming_blocked_components")[0]
assert record["value"] == "com.example.dialer,com.example.camera"
assert record["default"] == "com.android.settings,\n com.android.vending"
def test_trailing_metadata_is_not_part_of_the_value(self):
settings = parse_bugreport_settings()
record = find(settings, "lock_screen_show_notifications")[0]
assert record["value"] == "1"
assert record["defaultSystemSet"] == "true"
assert record["isValuePreservedInRestore"] == "true"
# Without a default, the tag or the restore token follows the value.
record = find(settings, "accessibility_enabled")[1]
assert record["value"] == "0"
assert record["tag"] == "null"
assert "default" not in record
record = find(settings, "send_action_app_error")[0]
assert record["value"] == "1"
assert record["isValuePreservedInRestore"] == "false"
def test_dumps_after_the_last_block_are_not_part_of_the_last_row(self):
settings = parse_bugreport_settings()
assert settings.results[-1] == {
"namespace": "secure",
"user": "10",
"_id": "311",
"name": "accessibility_enabled",
"pkg": "android",
"value": "0",
"tag": "null",
"history": [],
}
def test_repeated_names_are_kept_as_separate_records(self):
settings = parse_bugreport_settings()
widgets = find(settings, "widget_instance_data")
assert [record["_id"] for record in widgets] == ["771", "41654"]
accessibility = find(settings, "accessibility_enabled")
assert [(record["user"], record["value"]) for record in accessibility] == [
("0", "1"),
("10", "0"),
]
def test_setting_without_recording_package(self):
settings = parse_bugreport_settings()
record = find(settings, "hidden_api_blacklist_exemptions")[0]
assert "pkg" not in record
assert record["value"] == "{null}"
def test_history_timestamps_resolved_against_section_end(self):
settings = parse_bugreport_settings()
# The section was dumped on 2022-03-29, so an 11-02 entry belongs to
# the previous year and an 03-14 entry to the same year.
assert find(settings, "development_settings_enabled")[0]["history"] == [
{
"timestamp": "2021-11-02 11:21:22.212000",
"oldValue": "null",
"newValue": "1",
"pkg": "com.android.settings",
},
{
"timestamp": "2022-03-14 09:02:11.100000",
"oldValue": "1",
"newValue": "0",
"pkg": "com.example.updater",
},
]
def test_history_without_a_section_end_has_no_timestamp(self):
settings = Settings()
settings.parse(
"SECURE SETTINGS (user 0)\n"
"_id:240 name:accessibility_enabled pkg:android value:1\n"
"\tHistory (accessibility_enabled)\n"
"\t\ttime:03-28 22:41:07.980 mode:update oldValue:0 newValue:1 "
"package:com.example.helper\n"
)
assert settings.results[0]["history"] == [
{
"timestamp": None,
"oldValue": "0",
"newValue": "1",
"pkg": "com.example.helper",
}
]
def test_dangerous_setting_is_detected_with_the_changing_package(self):
settings = parse_bugreport_settings()
settings.check_indicators()
assert len(settings.alertstore.alerts) == 1
alert = settings.alertstore.alerts[0]
assert "accessibility_enabled = 1" in alert.message
assert alert.event_time == "2022-03-28 22:41:07.980000"
assert alert.event["history"][0]["pkg"] == "com.example.helper"
+8 -17
View File
@@ -6,27 +6,12 @@
from pathlib import Path
from mvt.android.modules.androidqf.aqf_settings import AQFSettings
from mvt.android.artifacts.settings import Settings
from mvt.common.module import run_module
from ..utils import get_android_androidqf, list_files
class TestSettingsModule:
def test_bugreport_settings_format(self):
settings = Settings()
settings.parse(
"GLOBAL SETTINGS (user 0)\n"
"_id:1 name:adb_wifi_enabled pkg:android value:0 default:0 defaultSystemSet:true\n"
"SECURE SETTINGS (user 10)\n"
"_id:2 name:accessibility_enabled pkg:android value:1\n"
)
assert settings.results == {
"global:user_0": {"adb_wifi_enabled": "0"},
"secure:user_10": {"accessibility_enabled": "1"},
}
def test_parsing(self):
data_path = get_android_androidqf()
m = AQFSettings(target_path=data_path)
@@ -34,7 +19,13 @@ class TestSettingsModule:
parent_path = Path(data_path).absolute().parent.as_posix()
m.from_dir(parent_path, files)
run_module(m)
assert len(m.results) == 1
assert "random" in m.results.keys()
assert len(m.results) == 9
assert {result["namespace"] for result in m.results} == {"random"}
assert m.results[0] == {
"namespace": "random",
"user": None,
"name": "samsung_errorlog_agree",
"value": "0",
}
assert len(m.alertstore.alerts) == 1
assert "samsung_errorlog_agree" in m.alertstore.alerts[0].message
+20
View File
@@ -10,6 +10,7 @@ from mvt.android.modules.bugreport.dumpsys_appops import DumpsysAppops
from mvt.android.modules.bugreport.dumpsys_getprop import DumpsysGetProp
from mvt.android.modules.bugreport.dumpsys_packages import DumpsysPackages
from mvt.android.modules.bugreport.dumpsys_receivers import DumpsysReceivers
from mvt.android.modules.bugreport.settings import Settings
from mvt.android.modules.bugreport.tombstones import Tombstones
from mvt.common.module import run_module
@@ -93,6 +94,25 @@ class TestBugreportAnalysis:
assert alert.event == malicious_receiver
assert alert.matched_indicator.value == "com.android.services"
def test_settings_module(self):
m = self.launch_bug_report_module(Settings)
assert len(m.results) == 12
assert len(m.alertstore.alerts) == 1
assert "accessibility_enabled = 1" in m.alertstore.alerts[0].message
assert len(m.timeline) == 3
change = [
entry
for entry in m.timeline
if entry["timestamp"] == "2022-03-28 22:41:07.980000"
][0]
assert change["event"] == "settings_change"
assert change["data"] == (
'secure setting "accessibility_enabled" changed from "0" to "1" '
"by com.example.helper"
)
def test_tombstones_modules(self):
m = self.launch_bug_report_module(Tombstones)
assert len(m.results) == 2
@@ -264,5 +264,53 @@ ChangeId(143539591; name=SELINUX_LATEST_CHANGES; disabled)
ChangeId(247079863; name=DISALLOW_INVALID_GROUP_REFERENCE; enableSinceTargetSdk=34)
ChangeId(174227820; name=FORCE_DISABLE_HEVC_SUPPORT; disabled)
ChangeId(168419799; name=DOWNSCALED; disabled; packageOverrides={com.google.android.apps.tachyon=false, org.torproject.torbrowser=false}; rawOverrides={org.torproject.torbrowser=false, org.article19.circulo.next=false}; overridable)
-------------------------------------------------------------------------------
DUMP OF SERVICE settings:
CONFIG SETTINGS (user 0)
_id:682 name:namespace_one/blocked_components pkg:com.example.services value:com.android.settings,com.android.vending,
com.example.dialer,
com.example.camera default:com.android.settings,
com.android.vending,
com.example.dialer defaultSystemSet:false
_id:684 name:namespace_one/streaming_blocked_components pkg:com.example.services value:com.example.dialer,com.example.camera default:com.android.settings,
com.android.vending defaultSystemSet:false
_id:680 name:namespace_one/allowed_packages pkg:com.example.services value:com.example.messaging,
com.example.chat
GLOBAL SETTINGS (user 0)
_id:2070 name:adb_wifi_enabled pkg:android value:0 default:0 defaultSystemSet:true
_id:778 name:hidden_api_blacklist_exemptions value:{null}
_id:9640 name:send_action_app_error pkg:android value:1 notPreservedInRestore
_id:9631 name:development_settings_enabled pkg:com.android.settings value:1 default:1 defaultSystemSet:true
History (development_settings_enabled)
time:11-02 11:21:22.212 mode:update oldValue:null newValue:1 package:com.android.settings
time:03-14 09:02:11.100 mode:update oldValue:1 newValue:0 package:com.example.updater
_id:771 name:widget_instance_data pkg:com.android.systemui value:{
"version": 1,
"data": [
{
"number": 10000,
"package_name": "com.example.widget"
}
]
} defaultSystemSet:true
_id:41654 name:widget_instance_data pkg:com.android.systemui value:{
"version": 3,
"data": []
} defaultSystemSet:true
SECURE SETTINGS (user 0)
_id:907 name:lock_screen_show_notifications pkg:com.android.settings value:1 default:1 defaultSystemSet:true isValuePreservedInRestore:true
_id:240 name:accessibility_enabled pkg:android value:1 default:0 defaultSystemSet:true
History (accessibility_enabled)
time:03-28 22:41:07.980 mode:update oldValue:0 newValue:1 package:com.example.helper
SECURE SETTINGS (user 10)
_id:311 name:accessibility_enabled pkg:android value:0 tag:null
GENERATION REGISTRY
Maximum number of backing stores:8
Number of backing stores:1
_Backing store for type:SETTINGS_SECURE user:10 size:1024 cachedEntries:1
--------- 0.019s was the duration of dumpsys settings, ending at: 2022-03-29 23:14:28
+100
View File
@@ -5,8 +5,13 @@
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
@@ -197,6 +202,86 @@ 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 = []
@@ -213,6 +298,21 @@ class TestCommand:
alerts = json.loads((tmp_path / "alerts.json").read_text())
assert alerts[0]["event"]["payload"] == "\\xa8\\xa9"
def test_records_time_and_memory_per_module(self, tmp_path):
cmd = RecordingCommand(results_path=str(tmp_path))
cmd.modules = [FirstModule, SecondModule]
cmd.run()
info = json.loads((tmp_path / "info.json").read_text())
assert [stat["module"] for stat in info["module_stats"]] == [
"FirstModule",
"SecondModule",
]
for stat in info["module_stats"]:
assert stat["seconds"] >= 0
assert stat["peak_rss_growth_mb"] >= 0
def test_stores_collected_urls(self, tmp_path):
cmd = RecordingCommand(results_path=str(tmp_path))
cmd.modules = [URLRecordingModule]
+4 -1
View File
@@ -12,13 +12,16 @@ 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
assert IOS_CHECK_IOCS_MODULES == (
IOS_BACKUP_MODULES + FS_MODULES + MIXED_MODULES + SYSDIAGNOSE_MODULES
)
assert ANDROID_CHECK_IOCS_MODULES == (
ANDROID_BACKUP_MODULES
+ BUGREPORT_MODULES
+30
View File
@@ -0,0 +1,30 @@
# 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 os
import yaml
from mvt.common import config
from mvt.common.config import MVTSettings
def test_env_variables_are_not_persisted_to_config_file(tmp_path, monkeypatch):
config_path = tmp_path / "config.yaml"
monkeypatch.setattr(config, "MVT_CONFIG_FOLDER", str(tmp_path))
monkeypatch.setattr(config, "MVT_CONFIG_PATH", str(config_path))
monkeypatch.setenv("MVT_NETWORK_ACCESS_ALLOWED", "false")
monkeypatch.setenv("MVT_IOS_BACKUP_PASSWORD", "env-only-password")
settings = MVTSettings.initialise()
assert os.path.isfile(config_path)
saved = yaml.safe_load(config_path.read_text()) or {}
assert "NETWORK_ACCESS_ALLOWED" not in saved
assert "IOS_BACKUP_PASSWORD" not in saved
# The environment must still apply to the settings in use.
assert settings.NETWORK_ACCESS_ALLOWED is False
assert settings.IOS_BACKUP_PASSWORD == "env-only-password"
+20 -7
View File
@@ -3,20 +3,26 @@
# 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 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
from .artifacts.generate_stix import generate_test_stix_file
# 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")
@pytest.fixture(scope="session", autouse=True)
def indicator_file(request, tmp_path_factory):
@@ -47,6 +53,8 @@ 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)
@@ -77,6 +85,11 @@ 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)
+4
View File
@@ -0,0 +1,4 @@
# 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/
@@ -0,0 +1,161 @@
# 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,11 +3,16 @@
# Use of this software is governed by the MVT License 1.1 that can be found at
# https://license.mvt.re/1.1/
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
@@ -28,3 +33,90 @@ 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,6 +19,14 @@ 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"
+39 -4
View File
@@ -1,3 +1,7 @@
import logging
import os
import tarfile
from click.testing import CliRunner
from mvt.ios.cli import check_sysdiagnose
@@ -50,8 +54,39 @@ def test_check_sysdiagnose_runs_explicitly_scoped_custom_module(tmp_path):
assert (output_path / "custom_sysdiagnose_module.json").exists()
def test_check_sysdiagnose_requires_an_explicitly_scoped_module(tmp_path):
result = CliRunner().invoke(check_sysdiagnose, [str(_create_sysdiagnose_folder(tmp_path))])
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))]
)
assert result.exit_code != 0
assert "No custom modules support mvt-ios check-sysdiagnose" in result.output
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
+13 -7
View File
@@ -34,6 +34,7 @@ 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
@@ -45,6 +46,11 @@ 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]
@@ -56,30 +62,30 @@ def _run_command(path):
def test_check_sysdiagnose_from_folder(tmp_path):
command = _run_command(_create_sysdiagnose_folder(tmp_path))
assert command.executed[0].results == [
assert _test_module(command).results == [
{"content": "artifact", "timezone_offset": timedelta(hours=2).seconds}
]
assert command.executed[0].ips_files == [
assert _test_module(command).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 command.executed[0].results == [
assert _test_module(command).results == [
{"content": "artifact", "timezone_offset": timedelta(hours=2).seconds}
]
assert command.executed[0].ips_files == [
assert _test_module(command).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):