diff --git a/src/mvt/android/cli.py b/src/mvt/android/cli.py index e9f8d019..4c159857 100644 --- a/src/mvt/android/cli.py +++ b/src/mvt/android/cli.py @@ -200,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: @@ -276,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() @@ -359,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() @@ -441,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() diff --git a/src/mvt/android/cmd_check_androidqf.py b/src/mvt/android/cmd_check_androidqf.py index 99ac5a3f..a86e643e 100644 --- a/src/mvt/android/cmd_check_androidqf.py +++ b/src/mvt/android/cmd_check_androidqf.py @@ -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() diff --git a/src/mvt/android/cmd_check_backup.py b/src/mvt/android/cmd_check_backup.py index 94be4ee2..066cd8aa 100644 --- a/src/mvt/android/cmd_check_backup.py +++ b/src/mvt/android/cmd_check_backup.py @@ -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" diff --git a/src/mvt/android/cmd_check_bugreport.py b/src/mvt/android/cmd_check_bugreport.py index 946feb97..036160d8 100644 --- a/src/mvt/android/cmd_check_bugreport.py +++ b/src/mvt/android/cmd_check_bugreport.py @@ -90,6 +90,7 @@ class CmdAndroidCheckBugreport(Command): def init(self) -> None: if self.target_path: + self.log.info("Checking Android bug report at path: %s", self.target_path) if os.path.isfile(self.target_path): self.from_zip(ZipFile(self.target_path)) elif os.path.isdir(self.target_path): diff --git a/src/mvt/android/cmd_check_intrusion_logs.py b/src/mvt/android/cmd_check_intrusion_logs.py index 95f20896..a652c229 100644 --- a/src/mvt/android/cmd_check_intrusion_logs.py +++ b/src/mvt/android/cmd_check_intrusion_logs.py @@ -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") diff --git a/src/mvt/common/cmd_check_iocs.py b/src/mvt/common/cmd_check_iocs.py index c2dbdcd9..7c7e71e5 100644 --- a/src/mvt/common/cmd_check_iocs.py +++ b/src/mvt/common/cmd_check_iocs.py @@ -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: diff --git a/src/mvt/common/command.py b/src/mvt/common/command.py index 12429417..6848e8c0 100644 --- a/src/mvt/common/command.py +++ b/src/mvt/common/command.py @@ -8,6 +8,7 @@ import logging import os import sys from datetime import datetime +from functools import cached_property from heapq import heappop, heappush from typing import Any, Optional @@ -84,18 +85,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: @@ -710,17 +711,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 +754,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: diff --git a/src/mvt/ios/cli.py b/src/mvt/ios/cli.py index 4560010a..f9f58885 100644 --- a/src/mvt/ios/cli.py +++ b/src/mvt/ios/cli.py @@ -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() diff --git a/src/mvt/ios/cmd_check_backup.py b/src/mvt/ios/cmd_check_backup.py index 1b90584b..42b13c56 100644 --- a/src/mvt/ios/cmd_check_backup.py +++ b/src/mvt/ios/cmd_check_backup.py @@ -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: diff --git a/src/mvt/ios/cmd_check_fs.py b/src/mvt/ios/cmd_check_fs.py index e76146ec..3daed974 100644 --- a/src/mvt/ios/cmd_check_fs.py +++ b/src/mvt/ios/cmd_check_fs.py @@ -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 diff --git a/src/mvt/ios/cmd_check_sysdiagnose.py b/src/mvt/ios/cmd_check_sysdiagnose.py index 88a12db0..c700e6bb 100644 --- a/src/mvt/ios/cmd_check_sysdiagnose.py +++ b/src/mvt/ios/cmd_check_sysdiagnose.py @@ -81,6 +81,8 @@ 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 diff --git a/tests/common/test_command.py b/tests/common/test_command.py index 8e09f57c..357d3b7d 100644 --- a/tests/common/test_command.py +++ b/tests/common/test_command.py @@ -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 = [] diff --git a/tests/test_check_ios_backup.py b/tests/test_check_ios_backup.py index eb10b131..3290e6d1 100644 --- a/tests/test_check_ios_backup.py +++ b/tests/test_check_ios_backup.py @@ -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"