mirror of
https://github.com/mvt-project/mvt.git
synced 2026-10-25 00:28:30 +02:00
Compare commits
6
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
06cea62a40 | ||
|
|
eceafdc785 | ||
|
|
ffae240355 | ||
|
|
3623e430f4 | ||
|
|
60e652bafc | ||
|
|
92f780df5a |
No files matched your search
+17
-8
@@ -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()
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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"
|
||||
),
|
||||
|
||||
@@ -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
@@ -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)
|
||||
|
||||
@@ -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")
|
||||
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
@@ -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():
|
||||
|
||||
@@ -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
@@ -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)
|
||||
|
||||
@@ -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"
|
||||
@@ -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"
|
||||
|
||||
@@ -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
|
||||
@@ -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):
|
||||
|
||||
Reference in new issue
Block a user