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
21 changed files with 466 additions and 83 deletions

No files matched your search

+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")
+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"
),
+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)
+3 -1
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")
+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)
-5
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()
@@ -463,7 +459,6 @@ def check_sysdiagnose(
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
+24 -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
@@ -79,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)
@@ -97,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
@@ -122,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():
+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]
+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)
+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"
+28
View File
@@ -1,4 +1,6 @@
import logging
import os
import tarfile
from click.testing import CliRunner
@@ -62,3 +64,29 @@ def test_check_sysdiagnose_warns_without_a_custom_module(tmp_path, caplog):
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
+3
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
@@ -67,6 +68,7 @@ def test_check_sysdiagnose_from_folder(tmp_path):
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):
@@ -83,6 +85,7 @@ def test_check_sysdiagnose_from_archive_closes_archive(tmp_path):
}
]
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):