diff --git a/docs/development.md b/docs/development.md index 93a612f..ffe1265 100644 --- a/docs/development.md +++ b/docs/development.md @@ -82,14 +82,16 @@ class ExampleCustomModule(MVTModule): return None ``` -Use `supported_commands` to restrict a module to specific platform/command -pairs. Missing or empty `supported_commands` means the module is available to -all commands, which keeps older modules compatible. Supported pairs are: +Use `supported_commands` to declare the platform/command pairs a module +supports. Empty `supported_commands` means the module will not run and MVT logs +a warning. This explicit declaration is required for every command. Supported +pairs are: ```python ("ios", "check-backup") ("ios", "check-fs") ("ios", "check-iocs") +("ios", "check-sysdiagnose") ("android", "check-backup") ("android", "check-bugreport") ("android", "check-androidqf") diff --git a/docs/ios/sysdiagnose.md b/docs/ios/sysdiagnose.md new file mode 100644 index 0000000..6488bc2 --- /dev/null +++ b/docs/ios/sysdiagnose.md @@ -0,0 +1,50 @@ +# 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. You must +load at least one custom module that explicitly supports this command. + +The command accepts either an extracted sysdiagnose directory or the original +gzip-compressed tar archive. + +```bash +mvt-ios check-sysdiagnose \ + --load-module ./sysdiagnose_modules.py \ + --output ./results \ + ./sysdiagnose_2024.01.02_03-04-05+0200.tar.gz +``` + +Use `--hashes` to include hashes for analyzed files in `info.json`, and +`--list-modules` to display the eligible custom modules without running them. + +## Writing a custom module + +Extend `SysdiagnoseExtraction` to access the archive contents consistently for +both directory and tar inputs. Each module must declare the command explicitly +in `supported_commands`. + +```python +from mvt.ios.modules.sysdiagnose import SysdiagnoseExtraction + + +class ExampleSysdiagnoseModule(SysdiagnoseExtraction): + supported_commands = (("ios", "check-sysdiagnose"),) + slug = "example_sysdiagnose" + + def run(self): + paths = self._get_files_by_pattern("*/example.log") + if paths: + content = self._get_file_content(paths[0]).decode("utf-8", "replace") + self.results = [{"content": content}] + + def check_indicators(self): + pass + + def serialize(self, result): + return None +``` + +The base class provides `from_sysdiagnose_folder()` and +`from_sysdiagnose_tar()` setup hooks, as well as protected file lookup, file +reading, and timezone extraction helpers. IPS crash-report metadata is exposed +on `ips_files`. diff --git a/mkdocs.yml b/mkdocs.yml index d34c3fa..3890b41 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -39,6 +39,7 @@ nav: - Check a Filesystem Dump: - Dumping the filesystem: "ios/filesystem/dump.md" - Check a Filesystem Dump with mvt-ios: "ios/filesystem/check.md" + - Check a Sysdiagnose: "ios/sysdiagnose.md" - Records extracted by mvt-ios: "ios/records.md" - MVT for Android: - Android Forensic Methodology: "android/methodology.md" diff --git a/src/mvt/common/help.py b/src/mvt/common/help.py index 90e71ca..5101f93 100644 --- a/src/mvt/common/help.py +++ b/src/mvt/common/help.py @@ -38,6 +38,7 @@ HELP_MSG_BACKUP_KEYFILE = ( HELP_MSG_EXTRACT_KEY = "Extract decryption key from an iTunes backup" HELP_MSG_CHECK_IOS_BACKUP = "Extract artifacts from an iTunes backup" HELP_MSG_CHECK_FS = "Extract artifacts from a full filesystem dump" +HELP_MSG_CHECK_SYSDIAGNOSE = "Extract artifacts from an iOS sysdiagnose archive" # Android Specific HELP_MSG_ANDROID_BACKUP_PASSWORD = "The backup password to use for an Android backup" diff --git a/src/mvt/common/module_loader.py b/src/mvt/common/module_loader.py index 890ed0f..da94826 100644 --- a/src/mvt/common/module_loader.py +++ b/src/mvt/common/module_loader.py @@ -6,6 +6,7 @@ import hashlib import importlib.util import inspect +import logging import os import sys from pathlib import Path @@ -15,6 +16,7 @@ from typing import Iterable, Optional from .module import MVTModule MVT_CUSTOM_MODULES_ENV = "MVT_CUSTOM_MODULES" +log = logging.getLogger(__name__) class CustomModuleLoadError(Exception): @@ -122,6 +124,11 @@ def module_supports_command( ) -> bool: supported_commands = getattr(module_class, "supported_commands", None) if not supported_commands: - return True + log.warning( + "Custom module %s has no supported_commands and will not be run. " + "Declare the platform/command pairs it supports.", + module_class.__name__, + ) + return False return (platform, command) in {tuple(entry) for entry in supported_commands} diff --git a/src/mvt/ios/cli.py b/src/mvt/ios/cli.py index 176b036..1e501f4 100644 --- a/src/mvt/ios/cli.py +++ b/src/mvt/ios/cli.py @@ -44,6 +44,7 @@ from mvt.common.help import ( HELP_MSG_CHECK_IOCS, HELP_MSG_STIX2, HELP_MSG_CHECK_IOS_BACKUP, + HELP_MSG_CHECK_SYSDIAGNOSE, HELP_MSG_DISABLE_UPDATE_CHECK, HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK, HELP_MSG_COMPLETION, @@ -51,6 +52,7 @@ from mvt.common.help import ( from mvt.common.module_loader import CustomModuleLoadError, load_custom_modules from .cmd_check_backup import CmdIOSCheckBackup from .cmd_check_fs import CmdIOSCheckFS +from .cmd_check_sysdiagnose import CmdIOSCheckSysdiagnose from .decrypt import DecryptBackup from .modules.backup import BACKUP_MODULES from .modules.fs import FS_MODULES @@ -396,6 +398,77 @@ def check_fs( cmd.show_support_message() +# ============================================================================== +# Command: check-sysdiagnose +# ============================================================================== +@cli.command( + "check-sysdiagnose", + context_settings=CONTEXT_SETTINGS, + help=HELP_MSG_CHECK_SYSDIAGNOSE, +) +@click.option( + "--iocs", + "-i", + type=click.Path(exists=True), + multiple=True, + default=[], + help=HELP_MSG_IOC, +) +@click.option("--output", "-o", type=click.Path(exists=False), help=HELP_MSG_OUTPUT) +@click.option("--list-modules", "-l", is_flag=True, help=HELP_MSG_LIST_MODULES) +@click.option("--module", "-m", help=HELP_MSG_MODULE) +@click.option( + "--load-module", + type=click.Path(exists=True), + multiple=True, + default=[], + help=HELP_MSG_LOAD_MODULE, +) +@click.option("--hashes", "-H", is_flag=True, help=HELP_MSG_HASHES) +@click.option("--verbose", "-v", is_flag=True, help=HELP_MSG_VERBOSE) +@click.argument("SYSDIAGNOSE_PATH", type=click.Path(exists=True)) +@click.pass_context +def check_sysdiagnose( + ctx, + iocs, + output, + list_modules, + module, + load_module, + hashes, + verbose, + sysdiagnose_path, +): + set_verbose_logging(verbose) + custom_modules = _load_custom_modules(load_module) + cmd = CmdIOSCheckSysdiagnose( + target_path=sysdiagnose_path, + results_path=output, + ioc_files=iocs, + module_name=module, + hashes=hashes, + disable_version_check=_get_disable_flags(ctx)[0], + disable_indicator_check=_get_disable_flags(ctx)[1], + 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\"),)." + ) + + 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() + + # ============================================================================== # Command: check-iocs # ============================================================================== diff --git a/src/mvt/ios/cmd_check_sysdiagnose.py b/src/mvt/ios/cmd_check_sysdiagnose.py new file mode 100644 index 0000000..6ceea6f --- /dev/null +++ b/src/mvt/ios/cmd_check_sysdiagnose.py @@ -0,0 +1,119 @@ +# 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/ + +import json +import logging +import os +import tarfile +from pathlib import Path +from typing import Any, Optional + +from mvt.common.command import Command +from mvt.common.indicators import Indicators +from mvt.common.module import MVTModule + +log = logging.getLogger(__name__) + + +class CmdIOSCheckSysdiagnose(Command): + def __init__( + self, + target_path: Optional[str] = None, + results_path: Optional[str] = None, + ioc_files: Optional[list] = None, + iocs: Optional[Indicators] = None, + module_name: Optional[str] = None, + serial: Optional[str] = None, + module_options: Optional[dict] = None, + hashes: bool = False, + sub_command: bool = False, + disable_version_check: bool = False, + disable_indicator_check: bool = False, + custom_modules: Optional[list[type[MVTModule]]] = None, + ) -> None: + super().__init__( + target_path=target_path, + results_path=results_path, + ioc_files=ioc_files, + iocs=iocs, + module_name=module_name, + serial=serial, + module_options=module_options, + hashes=hashes, + sub_command=sub_command, + log=log, + disable_version_check=disable_version_check, + disable_indicator_check=disable_indicator_check, + custom_modules=custom_modules, + ) + self.platform = "ios" + self.name = "check-sysdiagnose" + self.sysdiagnose_format: Optional[str] = None + self.sysdiagnose_archive: Optional[tarfile.TarFile] = None + self.sysdiagnose_files: list[str] = [] + self.ips_files: list[dict[str, Any]] = [] + + @staticmethod + def _parse_bugtype_header(data: bytes) -> Optional[int]: + try: + header = json.loads(data.split(b"\n", 1)[0].decode("utf-8")) + return int(header["bug_type"]) + except (json.JSONDecodeError, KeyError, UnicodeDecodeError, ValueError): + return None + + def _add_ips_file(self, file_path: str, data: bytes) -> None: + bug_type = self._parse_bugtype_header(data) + if bug_type is not None: + self.ips_files.append({"file_path": file_path, "bug_type": bug_type}) + + def init(self) -> None: + if not self.target_path: + raise ValueError("A sysdiagnose path is required") + + 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: + absolute_path = os.path.join(root, filename) + file_path = os.path.relpath(absolute_path, parent_path) + self.sysdiagnose_files.append(file_path) + if filename.endswith(".ips"): + with open(absolute_path, "rb") as handle: + self._add_ips_file(absolute_path, handle.read()) + return + + if not os.path.isfile(self.target_path): + raise ValueError(f"Sysdiagnose path does not exist: {self.target_path}") + + 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") + for member in self.sysdiagnose_archive: + self.sysdiagnose_files.append(member.name) + if member.isfile() and member.name.endswith(".ips"): + archive_handle = self.sysdiagnose_archive.extractfile(member) + if archive_handle is not None: + with archive_handle: + self._add_ips_file(member.name, archive_handle.read()) + + def module_init(self, module) -> None: + module.ips_files = self.ips_files + if self.sysdiagnose_format == "tar": + if self.sysdiagnose_archive is None: + raise RuntimeError("Sysdiagnose archive has not been initialized") + module.from_sysdiagnose_tar( + self.sysdiagnose_archive, self.sysdiagnose_files + ) + return + if self.sysdiagnose_format == "dir" and self.target_path: + module.from_sysdiagnose_folder(self.target_path, self.sysdiagnose_files) + return + raise RuntimeError("Sysdiagnose input has not been initialized") + + def finish(self) -> None: + if self.sysdiagnose_archive is not None: + self.sysdiagnose_archive.close() + self.sysdiagnose_archive = None diff --git a/src/mvt/ios/modules/sysdiagnose/__init__.py b/src/mvt/ios/modules/sysdiagnose/__init__.py new file mode 100644 index 0000000..3963ca5 --- /dev/null +++ b/src/mvt/ios/modules/sysdiagnose/__init__.py @@ -0,0 +1,6 @@ +# 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 .base import SysdiagnoseExtraction diff --git a/src/mvt/ios/modules/sysdiagnose/base.py b/src/mvt/ios/modules/sysdiagnose/base.py new file mode 100644 index 0000000..99187c9 --- /dev/null +++ b/src/mvt/ios/modules/sysdiagnose/base.py @@ -0,0 +1,105 @@ +# 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/ + +import fnmatch +import logging +import os +import re +import tarfile +from datetime import datetime, timezone +from pathlib import Path +from typing import Optional + +from mvt.common.module import MVTModule, ModuleResults + + +class SysdiagnoseExtraction(MVTModule): + """Base class for custom modules that analyze an iOS sysdiagnose.""" + + 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.parent_path: Optional[str] = None + self.files: list[str] = [] + self.tar: Optional[tarfile.TarFile] = None + self.tar_files: list[str] = [] + self.ips_files: list[dict[str, object]] = [] + + def from_sysdiagnose_folder( + self, target_path: str, sysdiagnose_files: list[str] + ) -> None: + self.parent_path = Path(target_path).absolute().parent.as_posix() + self.files = sysdiagnose_files + + def from_sysdiagnose_tar( + self, sysdiagnose_archive: tarfile.TarFile, sysdiagnose_files: list[str] + ) -> None: + self.tar = sysdiagnose_archive + self.tar_files = sysdiagnose_files + + def _extract_timezone(self): + """Determine the sysdiagnose timezone from its diagnostic log.""" + file_paths = self._get_files_by_pattern("*/sysdiagnose.log") + if not file_paths: + self.log.info( + "Unable to determine the timezone in which the sysdiagnose was " + "generated. Assuming UTC for logs without a timezone." + ) + return timezone.utc + + content = self._get_file_content(file_paths[0]).decode( + "utf-8", errors="replace" + ) + filenames = re.findall(r"sysdiagnose_\S+?\.tar\.gz", content) + if not filenames: + self.log.info( + "Unable to determine the timezone in which the sysdiagnose was " + "generated. Assuming UTC for logs without a timezone." + ) + return timezone.utc + + timestamp = "_".join( + filenames[0].removesuffix(".tar.gz").split("_")[1:3] + ) + sysdiagnose_timezone = datetime.strptime( + timestamp, "%Y.%m.%d_%H-%M-%S%z" + ).tzinfo + self.log.info( + "Based on the sysdiagnose filename, assuming timezone %s for logs " + "without a timezone.", + sysdiagnose_timezone, + ) + return sysdiagnose_timezone + + def _get_files_by_pattern(self, pattern: str) -> list[str]: + file_names = self.tar_files if self.tar else self.files + return fnmatch.filter(file_names, pattern) + + def _get_file_content(self, file_path: str) -> bytes: + if self.tar: + handle = self.tar.extractfile(self.tar.getmember(file_path)) + else: + if self.parent_path is None: + raise RuntimeError("Sysdiagnose folder has not been initialized") + handle = open(os.path.join(self.parent_path, file_path), "rb") + + if handle is None: + return b"" + with handle: + return handle.read() diff --git a/tests/common/test_command.py b/tests/common/test_command.py index 865f9f1..fa90a3e 100644 --- a/tests/common/test_command.py +++ b/tests/common/test_command.py @@ -50,6 +50,10 @@ class CustomIOSFSModule(RecordingModule): supported_commands = (("ios", "check-fs"),) +class UnscopedCustomModule(RecordingModule): + pass + + class CustomDependsOnBuiltin(RecordingModule): supported_commands = (("ios", "check-backup"),) dependencies = (FirstModule,) @@ -151,7 +155,11 @@ class TestCommand: cmd.platform = "ios" cmd.name = "check-backup" cmd.modules = [FirstModule] - cmd.custom_modules = [CustomIOSBackupModule, CustomIOSFSModule] + cmd.custom_modules = [ + CustomIOSBackupModule, + CustomIOSFSModule, + UnscopedCustomModule, + ] assert [module.__name__ for module in cmd._ordered_modules()] == [ "FirstModule", diff --git a/tests/common/test_module_loader.py b/tests/common/test_module_loader.py index ce13427..15b6052 100644 --- a/tests/common/test_module_loader.py +++ b/tests/common/test_module_loader.py @@ -125,12 +125,13 @@ def test_load_custom_modules_loads_env_folder_first(tmp_path, monkeypatch): assert [module.__name__ for module in modules] == ["EnvModule", "CliModule"] -def test_module_supports_command_defaults_to_all_commands(tmp_path): +def test_module_supports_command_requires_explicit_declaration(tmp_path, caplog): module_path = _write_module(tmp_path / "custom.py", "DefaultModule") module = load_custom_modules_from_path(str(module_path))[0] - assert module_supports_command(module, "ios", "check-backup") - assert module_supports_command(module, "android", "check-bugreport") + assert not module_supports_command(module, "ios", "check-backup") + assert not module_supports_command(module, "android", "check-bugreport") + assert "DefaultModule has no supported_commands" in caplog.text def test_module_supports_command_honors_supported_commands(tmp_path): diff --git a/tests/test_check_ios_sysdiagnose.py b/tests/test_check_ios_sysdiagnose.py new file mode 100644 index 0000000..d73c0ff --- /dev/null +++ b/tests/test_check_ios_sysdiagnose.py @@ -0,0 +1,57 @@ +from click.testing import CliRunner + +from mvt.ios.cli import check_sysdiagnose + + +CUSTOM_MODULE = """ +from mvt.ios.modules.sysdiagnose import SysdiagnoseExtraction + + +class CustomSysdiagnoseModule(SysdiagnoseExtraction): + supported_commands = (("ios", "check-sysdiagnose"),) + slug = "custom_sysdiagnose_module" + + def run(self): + file_path = self._get_files_by_pattern("*/artifact.txt")[0] + self.results = [{"content": self._get_file_content(file_path).decode("utf-8")}] + + def check_indicators(self): + pass + + def serialize(self, result): + return None +""" + + +def _create_sysdiagnose_folder(tmp_path): + folder = tmp_path / "sysdiagnose" + folder.mkdir() + (folder / "artifact.txt").write_text("artifact", encoding="utf-8") + return folder + + +def test_check_sysdiagnose_runs_explicitly_scoped_custom_module(tmp_path): + module_path = tmp_path / "custom_sysdiagnose.py" + module_path.write_text(CUSTOM_MODULE, encoding="utf-8") + output_path = tmp_path / "output" + + result = CliRunner().invoke( + check_sysdiagnose, + [ + "--load-module", + str(module_path), + "--output", + str(output_path), + str(_create_sysdiagnose_folder(tmp_path)), + ], + ) + + assert result.exit_code == 0 + 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))]) + + assert result.exit_code != 0 + assert "No custom modules support mvt-ios check-sysdiagnose" in result.output diff --git a/tests/test_cmd_check_sysdiagnose.py b/tests/test_cmd_check_sysdiagnose.py new file mode 100644 index 0000000..9f792de --- /dev/null +++ b/tests/test_cmd_check_sysdiagnose.py @@ -0,0 +1,75 @@ +import tarfile +from datetime import timedelta + +from mvt.ios.cmd_check_sysdiagnose import CmdIOSCheckSysdiagnose +from mvt.ios.modules.sysdiagnose import SysdiagnoseExtraction + + +class SysdiagnoseTestModule(SysdiagnoseExtraction): + supported_commands = (("ios", "check-sysdiagnose"),) + + def run(self): + file_path = self._get_files_by_pattern("*/artifact.txt")[0] + self.results = [ + { + "content": self._get_file_content(file_path).decode("utf-8"), + "timezone_offset": self._extract_timezone().utcoffset(None).seconds, + } + ] + + def check_indicators(self): + pass + + def serialize(self, result): + return None + + +def _create_sysdiagnose_folder(tmp_path): + folder = tmp_path / "sysdiagnose" + folder.mkdir() + (folder / "artifact.txt").write_text("artifact", encoding="utf-8") + (folder / "sysdiagnose.log").write_text( + "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") + return folder + + +def _create_sysdiagnose_archive(tmp_path, folder): + archive_path = tmp_path / "sysdiagnose.tar.gz" + with tarfile.open(archive_path, "w:gz") as archive: + for path in folder.iterdir(): + archive.add(path, arcname=f"sysdiagnose/{path.name}") + return archive_path + + +def _run_command(path): + command = CmdIOSCheckSysdiagnose( + target_path=str(path), custom_modules=[SysdiagnoseTestModule] + ) + command.run() + return command + + +def test_check_sysdiagnose_from_folder(tmp_path): + command = _run_command(_create_sysdiagnose_folder(tmp_path)) + + assert command.executed[0].results == [ + {"content": "artifact", "timezone_offset": timedelta(hours=2).seconds} + ] + assert command.executed[0].ips_files == [ + {"file_path": str(tmp_path / "sysdiagnose" / "report.ips"), "bug_type": 210} + ] + + +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 == [ + {"content": "artifact", "timezone_offset": timedelta(hours=2).seconds} + ] + assert command.executed[0].ips_files == [ + {"file_path": "sysdiagnose/report.ips", "bug_type": 210} + ] + assert command.sysdiagnose_archive is None