mirror of
https://github.com/mvt-project/mvt.git
synced 2026-08-17 16:37:23 +02:00
Add iOS sysdiagnose checking (#832)
* Add iOS sysdiagnose checking * Clarify documentation navigation
This commit is contained in:
+5
-3
@@ -82,14 +82,16 @@ class ExampleCustomModule(MVTModule):
|
|||||||
return None
|
return None
|
||||||
```
|
```
|
||||||
|
|
||||||
Use `supported_commands` to restrict a module to specific platform/command
|
Use `supported_commands` to declare the platform/command pairs a module
|
||||||
pairs. Missing or empty `supported_commands` means the module is available to
|
supports. Empty `supported_commands` means the module will not run and MVT logs
|
||||||
all commands, which keeps older modules compatible. Supported pairs are:
|
a warning. This explicit declaration is required for every command. Supported
|
||||||
|
pairs are:
|
||||||
|
|
||||||
```python
|
```python
|
||||||
("ios", "check-backup")
|
("ios", "check-backup")
|
||||||
("ios", "check-fs")
|
("ios", "check-fs")
|
||||||
("ios", "check-iocs")
|
("ios", "check-iocs")
|
||||||
|
("ios", "check-sysdiagnose")
|
||||||
("android", "check-backup")
|
("android", "check-backup")
|
||||||
("android", "check-bugreport")
|
("android", "check-bugreport")
|
||||||
("android", "check-androidqf")
|
("android", "check-androidqf")
|
||||||
|
|||||||
@@ -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`.
|
||||||
@@ -3,6 +3,8 @@ repo_url: https://github.com/mvt-project/mvt
|
|||||||
edit_uri: edit/main/docs/
|
edit_uri: edit/main/docs/
|
||||||
copyright: Copyright © 2021-2023 MVT Project Developers
|
copyright: Copyright © 2021-2023 MVT Project Developers
|
||||||
site_description: Mobile Verification Toolkit Documentation
|
site_description: Mobile Verification Toolkit Documentation
|
||||||
|
not_in_nav: |
|
||||||
|
android/adb.md
|
||||||
markdown_extensions:
|
markdown_extensions:
|
||||||
- attr_list
|
- attr_list
|
||||||
- admonition
|
- admonition
|
||||||
@@ -28,6 +30,7 @@ nav:
|
|||||||
- Welcome: "index.md"
|
- Welcome: "index.md"
|
||||||
- Introduction: "introduction.md"
|
- Introduction: "introduction.md"
|
||||||
- Installation: "install.md"
|
- Installation: "install.md"
|
||||||
|
- Command Completion: "command_completion.md"
|
||||||
- Using Docker: "docker.md"
|
- Using Docker: "docker.md"
|
||||||
- MVT for iOS:
|
- MVT for iOS:
|
||||||
- iOS Forensic Methodology: "ios/methodology.md"
|
- iOS Forensic Methodology: "ios/methodology.md"
|
||||||
@@ -39,6 +42,7 @@ nav:
|
|||||||
- Check a Filesystem Dump:
|
- Check a Filesystem Dump:
|
||||||
- Dumping the filesystem: "ios/filesystem/dump.md"
|
- Dumping the filesystem: "ios/filesystem/dump.md"
|
||||||
- Check a Filesystem Dump with mvt-ios: "ios/filesystem/check.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"
|
- Records extracted by mvt-ios: "ios/records.md"
|
||||||
- MVT for Android:
|
- MVT for Android:
|
||||||
- Android Forensic Methodology: "android/methodology.md"
|
- Android Forensic Methodology: "android/methodology.md"
|
||||||
|
|||||||
@@ -38,6 +38,7 @@ HELP_MSG_BACKUP_KEYFILE = (
|
|||||||
HELP_MSG_EXTRACT_KEY = "Extract decryption key from an iTunes backup"
|
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_IOS_BACKUP = "Extract artifacts from an iTunes backup"
|
||||||
HELP_MSG_CHECK_FS = "Extract artifacts from a full filesystem dump"
|
HELP_MSG_CHECK_FS = "Extract artifacts from a full filesystem dump"
|
||||||
|
HELP_MSG_CHECK_SYSDIAGNOSE = "Extract artifacts from an iOS sysdiagnose archive"
|
||||||
|
|
||||||
# Android Specific
|
# Android Specific
|
||||||
HELP_MSG_ANDROID_BACKUP_PASSWORD = "The backup password to use for an Android backup"
|
HELP_MSG_ANDROID_BACKUP_PASSWORD = "The backup password to use for an Android backup"
|
||||||
|
|||||||
@@ -6,6 +6,7 @@
|
|||||||
import hashlib
|
import hashlib
|
||||||
import importlib.util
|
import importlib.util
|
||||||
import inspect
|
import inspect
|
||||||
|
import logging
|
||||||
import os
|
import os
|
||||||
import sys
|
import sys
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
@@ -15,6 +16,7 @@ from typing import Iterable, Optional
|
|||||||
from .module import MVTModule
|
from .module import MVTModule
|
||||||
|
|
||||||
MVT_CUSTOM_MODULES_ENV = "MVT_CUSTOM_MODULES"
|
MVT_CUSTOM_MODULES_ENV = "MVT_CUSTOM_MODULES"
|
||||||
|
log = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
class CustomModuleLoadError(Exception):
|
class CustomModuleLoadError(Exception):
|
||||||
@@ -122,6 +124,11 @@ def module_supports_command(
|
|||||||
) -> bool:
|
) -> bool:
|
||||||
supported_commands = getattr(module_class, "supported_commands", None)
|
supported_commands = getattr(module_class, "supported_commands", None)
|
||||||
if not supported_commands:
|
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}
|
return (platform, command) in {tuple(entry) for entry in supported_commands}
|
||||||
|
|||||||
@@ -42,6 +42,7 @@ from mvt.common.help import (
|
|||||||
HELP_MSG_CHECK_IOCS,
|
HELP_MSG_CHECK_IOCS,
|
||||||
HELP_MSG_STIX2,
|
HELP_MSG_STIX2,
|
||||||
HELP_MSG_CHECK_IOS_BACKUP,
|
HELP_MSG_CHECK_IOS_BACKUP,
|
||||||
|
HELP_MSG_CHECK_SYSDIAGNOSE,
|
||||||
HELP_MSG_DISABLE_UPDATE_CHECK,
|
HELP_MSG_DISABLE_UPDATE_CHECK,
|
||||||
HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK,
|
HELP_MSG_DISABLE_INDICATOR_UPDATE_CHECK,
|
||||||
HELP_MSG_COMPLETION,
|
HELP_MSG_COMPLETION,
|
||||||
@@ -50,6 +51,7 @@ from mvt.common.module_loader import CustomModuleLoadError, load_custom_modules
|
|||||||
from mvt.common.password import prompt_password
|
from mvt.common.password import prompt_password
|
||||||
from .cmd_check_backup import CmdIOSCheckBackup
|
from .cmd_check_backup import CmdIOSCheckBackup
|
||||||
from .cmd_check_fs import CmdIOSCheckFS
|
from .cmd_check_fs import CmdIOSCheckFS
|
||||||
|
from .cmd_check_sysdiagnose import CmdIOSCheckSysdiagnose
|
||||||
from .decrypt import DecryptBackup
|
from .decrypt import DecryptBackup
|
||||||
from .modules.backup import BACKUP_MODULES
|
from .modules.backup import BACKUP_MODULES
|
||||||
from .modules.fs import FS_MODULES
|
from .modules.fs import FS_MODULES
|
||||||
@@ -398,6 +400,77 @@ def check_fs(
|
|||||||
cmd.show_support_message()
|
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
|
# Command: check-iocs
|
||||||
# ==============================================================================
|
# ==============================================================================
|
||||||
|
|||||||
@@ -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
|
||||||
@@ -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
|
||||||
@@ -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()
|
||||||
@@ -50,6 +50,10 @@ class CustomIOSFSModule(RecordingModule):
|
|||||||
supported_commands = (("ios", "check-fs"),)
|
supported_commands = (("ios", "check-fs"),)
|
||||||
|
|
||||||
|
|
||||||
|
class UnscopedCustomModule(RecordingModule):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
class CustomDependsOnBuiltin(RecordingModule):
|
class CustomDependsOnBuiltin(RecordingModule):
|
||||||
supported_commands = (("ios", "check-backup"),)
|
supported_commands = (("ios", "check-backup"),)
|
||||||
dependencies = (FirstModule,)
|
dependencies = (FirstModule,)
|
||||||
@@ -151,7 +155,11 @@ class TestCommand:
|
|||||||
cmd.platform = "ios"
|
cmd.platform = "ios"
|
||||||
cmd.name = "check-backup"
|
cmd.name = "check-backup"
|
||||||
cmd.modules = [FirstModule]
|
cmd.modules = [FirstModule]
|
||||||
cmd.custom_modules = [CustomIOSBackupModule, CustomIOSFSModule]
|
cmd.custom_modules = [
|
||||||
|
CustomIOSBackupModule,
|
||||||
|
CustomIOSFSModule,
|
||||||
|
UnscopedCustomModule,
|
||||||
|
]
|
||||||
|
|
||||||
assert [module.__name__ for module in cmd._ordered_modules()] == [
|
assert [module.__name__ for module in cmd._ordered_modules()] == [
|
||||||
"FirstModule",
|
"FirstModule",
|
||||||
|
|||||||
@@ -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"]
|
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_path = _write_module(tmp_path / "custom.py", "DefaultModule")
|
||||||
module = load_custom_modules_from_path(str(module_path))[0]
|
module = load_custom_modules_from_path(str(module_path))[0]
|
||||||
|
|
||||||
assert module_supports_command(module, "ios", "check-backup")
|
assert not module_supports_command(module, "ios", "check-backup")
|
||||||
assert module_supports_command(module, "android", "check-bugreport")
|
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):
|
def test_module_supports_command_honors_supported_commands(tmp_path):
|
||||||
|
|||||||
@@ -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
|
||||||
@@ -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
|
||||||
Reference in New Issue
Block a user