mirror of
https://github.com/mvt-project/mvt.git
synced 2026-10-25 00:28:30 +02:00
Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
4e031cd43f | ||
|
|
dac4acb180 | ||
|
|
dd8bd2cb01 |
No files matched your search
@@ -84,7 +84,7 @@ details.
|
|||||||
|
|
||||||
Users can also add top-level commands to `mvt-ios` and `mvt-android` from
|
Users can also add top-level commands to `mvt-ios` and `mvt-android` from
|
||||||
installed Python packages or local files and folders. See the
|
installed Python packages or local files and folders. See the
|
||||||
[custom CLI command documentation](https://docs.mvt.re/en/latest/custom_commands/)
|
[custom CLI command documentation](https://docs.mvt.re/en/latest/development/custom_commands/)
|
||||||
for the plugin entry points and `--load-command` interface.
|
for the plugin entry points and `--load-command` interface.
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -1,7 +1,7 @@
|
|||||||
# Custom CLI Commands
|
# Custom CLI Commands
|
||||||
|
|
||||||
MVT can load additional top-level commands into `mvt-ios` and `mvt-android`.
|
MVT can load additional top-level commands into `mvt-ios` and `mvt-android`.
|
||||||
Custom commands are different from [custom forensic modules](development.md#custom-modules):
|
Custom commands are different from [custom forensic modules](index.md#custom-modules):
|
||||||
commands add new CLI operations, while modules add analysis steps to existing
|
commands add new CLI operations, while modules add analysis steps to existing
|
||||||
`check-*` commands.
|
`check-*` commands.
|
||||||
|
|
||||||
@@ -117,6 +117,93 @@ class DependentCustomModule(MVTModule):
|
|||||||
self.results = [{"manifest_entries": len(manifest_results)}]
|
self.results = [{"manifest_entries": len(manifest_results)}]
|
||||||
```
|
```
|
||||||
|
|
||||||
|
## Installed module packages
|
||||||
|
|
||||||
|
Python packages can register modules so they load automatically in every
|
||||||
|
module-running `check-*` command, without `--load-module` or
|
||||||
|
`MVT_CUSTOM_MODULES`. Register an entry point in the `mvt.modules` group in
|
||||||
|
the package's `pyproject.toml`:
|
||||||
|
|
||||||
|
```toml
|
||||||
|
[project.entry-points."mvt.modules"]
|
||||||
|
mvt-plugin-amnesty-custom = "mvt_plugin_amnesty_custom:get_modules"
|
||||||
|
```
|
||||||
|
|
||||||
|
The entry point must resolve to an iterable of `MVTModule` subclasses, or to
|
||||||
|
a callable returning one:
|
||||||
|
|
||||||
|
```python
|
||||||
|
from mvt.common.module import MVTModule
|
||||||
|
|
||||||
|
|
||||||
|
class PackagedModule(MVTModule):
|
||||||
|
supported_commands = (("ios", "check-backup"),)
|
||||||
|
|
||||||
|
def run(self):
|
||||||
|
self.results = [{"message": "packaged module ran"}]
|
||||||
|
|
||||||
|
|
||||||
|
def get_modules() -> list[type[MVTModule]]:
|
||||||
|
return [PackagedModule]
|
||||||
|
```
|
||||||
|
|
||||||
|
Installed modules follow the same rules as other custom modules: each module
|
||||||
|
must declare `supported_commands`, and dependencies are resolved with the
|
||||||
|
standard ordering logic. A broken entry point is skipped with a warning and
|
||||||
|
does not prevent MVT from running. As with custom commands, installed module
|
||||||
|
packages run as trusted code inside the MVT process, so install only packages
|
||||||
|
from sources you trust.
|
||||||
|
|
||||||
|
For a `pipx` installation of MVT, inject the package into MVT's environment:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
pipx inject mvt mvt-plugin-amnesty-custom
|
||||||
|
```
|
||||||
|
|
||||||
|
### Naming module packages
|
||||||
|
|
||||||
|
Name module packages `mvt-plugin-<name>` (import package `mvt_plugin_<name>`),
|
||||||
|
and include the name of the publishing organization or author so packages from
|
||||||
|
different groups do not collide: for example, Amnesty International's custom
|
||||||
|
modules would be distributed as `mvt-plugin-amnesty-custom` with the import
|
||||||
|
package `mvt_plugin_amnesty_custom`.
|
||||||
|
|
||||||
|
The prefix makes module packages easy to find on PyPI and keeps their import
|
||||||
|
names from clashing with unrelated Python packages. It is a convention, not a
|
||||||
|
technical requirement: modules load through the `mvt.modules` entry point
|
||||||
|
regardless of what the package is called, and existing packages with other
|
||||||
|
names keep working. Note that the prefix is also not a mark of authenticity —
|
||||||
|
anyone can publish a package with any available name, so vet a module package
|
||||||
|
and its publisher before installing it, whatever it is called.
|
||||||
|
|
||||||
|
### Module logging
|
||||||
|
|
||||||
|
Modules log through `self.log`, and MVT names the logger for where the module
|
||||||
|
came from. MVT's own modules log under their dotted path (for example
|
||||||
|
`mvt.ios.modules.mixed.whatsapp`). Everything external is namespaced under
|
||||||
|
`mvt.ext` to keep it visually distinct from built-in modules and isolated from
|
||||||
|
MVT's internal logger tree:
|
||||||
|
|
||||||
|
- Installed packages log under `mvt.ext.<package>`, with the `mvt_plugin_`
|
||||||
|
prefix stripped: modules in `mvt_plugin_amnesty_custom` log as
|
||||||
|
`mvt.ext.amnesty_custom.*`.
|
||||||
|
- Files loaded with `--load-module` or `MVT_CUSTOM_MODULES` log as
|
||||||
|
`mvt.ext.<file name>`.
|
||||||
|
|
||||||
|
## Auditing loaded modules
|
||||||
|
|
||||||
|
Because installed module packages load automatically, MVT records where every
|
||||||
|
module came from:
|
||||||
|
|
||||||
|
- `--list-modules` groups the available modules by source: MVT itself
|
||||||
|
(with its version), each installed package (with its version and, when
|
||||||
|
installed directly from a repository, the commit), and each file loaded
|
||||||
|
with `--load-module` or `MVT_CUSTOM_MODULES` (with the SHA-256 hash of the
|
||||||
|
file).
|
||||||
|
- When a command runs with an `--output` folder, the `command.log` file
|
||||||
|
records one line per module source with the source's version or hash and
|
||||||
|
the list of modules loaded from it.
|
||||||
|
|
||||||
## Profiling
|
## Profiling
|
||||||
|
|
||||||
Some MVT modules extract and process significant amounts of data during the analysis process or while checking results against known indicators. Care must be
|
Some MVT modules extract and process significant amounts of data during the analysis process or while checking results against known indicators. Care must be
|
||||||
+3
-2
@@ -31,7 +31,6 @@ nav:
|
|||||||
- Introduction: "introduction.md"
|
- Introduction: "introduction.md"
|
||||||
- Installation: "install.md"
|
- Installation: "install.md"
|
||||||
- Command Completion: "command_completion.md"
|
- Command Completion: "command_completion.md"
|
||||||
- Custom CLI Commands: "custom_commands.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"
|
||||||
@@ -50,5 +49,7 @@ nav:
|
|||||||
- Check an Android Backup (SMS messages): "android/backup.md"
|
- Check an Android Backup (SMS messages): "android/backup.md"
|
||||||
- Check Android Intrusion Logs: "android/intrusion_logs.md"
|
- Check Android Intrusion Logs: "android/intrusion_logs.md"
|
||||||
- Indicators of Compromise: "iocs.md"
|
- Indicators of Compromise: "iocs.md"
|
||||||
- Development: "development.md"
|
- Development:
|
||||||
|
- Development Instructions: "development/index.md"
|
||||||
|
- Custom CLI Commands: "development/custom_commands.md"
|
||||||
- License: "license.md"
|
- License: "license.md"
|
||||||
@@ -9,6 +9,7 @@ from typing import Optional
|
|||||||
|
|
||||||
from mvt.common.command import Command
|
from mvt.common.command import Command
|
||||||
from mvt.common.module import MVTModule
|
from mvt.common.module import MVTModule
|
||||||
|
from mvt.common.module_loader import get_module_logger
|
||||||
from mvt.common.utils import exec_or_profile
|
from mvt.common.utils import exec_or_profile
|
||||||
|
|
||||||
log = logging.getLogger(__name__)
|
log = logging.getLogger(__name__)
|
||||||
@@ -76,7 +77,7 @@ class CmdCheckIOCS(Command):
|
|||||||
)
|
)
|
||||||
|
|
||||||
m = iocs_module.from_json(
|
m = iocs_module.from_json(
|
||||||
file_path, log=logging.getLogger(iocs_module.__module__)
|
file_path, log=get_module_logger(iocs_module)
|
||||||
)
|
)
|
||||||
if not m:
|
if not m:
|
||||||
log.warning("No result from this module, skipping it")
|
log.warning("No result from this module, skipping it")
|
||||||
|
|||||||
@@ -19,7 +19,12 @@ from .alerts import AlertLevel, AlertStore
|
|||||||
from .config import settings
|
from .config import settings
|
||||||
from .indicators import Indicators
|
from .indicators import Indicators
|
||||||
from .module import EncryptedBackupError, MVTModule, run_module, save_timeline
|
from .module import EncryptedBackupError, MVTModule, run_module, save_timeline
|
||||||
from .module_loader import module_supports_command
|
from .module_loader import (
|
||||||
|
ModuleOrigin,
|
||||||
|
get_module_logger,
|
||||||
|
get_module_origin,
|
||||||
|
module_supports_command,
|
||||||
|
)
|
||||||
from .module_types import ModuleTimeline, URLResult
|
from .module_types import ModuleTimeline, URLResult
|
||||||
from .utils import (
|
from .utils import (
|
||||||
CustomJSONEncoder,
|
CustomJSONEncoder,
|
||||||
@@ -210,10 +215,34 @@ class Command:
|
|||||||
for file in generate_hashes_from_path(self.target_path, self.log):
|
for file in generate_hashes_from_path(self.target_path, self.log):
|
||||||
self.hash_values.append(file)
|
self.hash_values.append(file)
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _modules_by_origin(
|
||||||
|
modules: list[type[MVTModule]],
|
||||||
|
) -> dict[ModuleOrigin, list[str]]:
|
||||||
|
origins: dict[ModuleOrigin, list[str]] = {}
|
||||||
|
for module in modules:
|
||||||
|
origins.setdefault(get_module_origin(module), []).append(module.__name__)
|
||||||
|
return origins
|
||||||
|
|
||||||
def list_modules(self) -> None:
|
def list_modules(self) -> None:
|
||||||
self.log.info("Following is the list of available %s modules:", self.name)
|
self.log.info("Following is the list of available %s modules:", self.name)
|
||||||
for module in self._available_modules():
|
for origin, module_names in self._modules_by_origin(
|
||||||
self.log.info(" - %s", module.__name__)
|
self._available_modules()
|
||||||
|
).items():
|
||||||
|
self.log.info(
|
||||||
|
" - Modules from %s: %s", origin.label, ", ".join(module_names)
|
||||||
|
)
|
||||||
|
|
||||||
|
def _log_loaded_modules(self, modules: list[type[MVTModule]]) -> None:
|
||||||
|
"""Record the loaded modules and their origins for auditability."""
|
||||||
|
for origin, module_names in self._modules_by_origin(modules).items():
|
||||||
|
self.log.info(
|
||||||
|
"Loaded %d %s modules from %s: %s",
|
||||||
|
len(module_names),
|
||||||
|
self.name,
|
||||||
|
origin.label,
|
||||||
|
", ".join(module_names),
|
||||||
|
)
|
||||||
|
|
||||||
def _available_modules(self) -> list[type[MVTModule]]:
|
def _available_modules(self) -> list[type[MVTModule]]:
|
||||||
modules = list(self.modules)
|
modules = list(self.modules)
|
||||||
@@ -360,6 +389,8 @@ class Command:
|
|||||||
if ordered_modules is None:
|
if ordered_modules is None:
|
||||||
return
|
return
|
||||||
|
|
||||||
|
self._log_loaded_modules(ordered_modules)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
self.init()
|
self.init()
|
||||||
except NotImplementedError:
|
except NotImplementedError:
|
||||||
@@ -368,7 +399,7 @@ class Command:
|
|||||||
executed_by_type: dict[type[MVTModule], MVTModule] = {}
|
executed_by_type: dict[type[MVTModule], MVTModule] = {}
|
||||||
for module in ordered_modules:
|
for module in ordered_modules:
|
||||||
|
|
||||||
module_logger = logging.getLogger(module.__module__)
|
module_logger = get_module_logger(module)
|
||||||
|
|
||||||
m = module(
|
m = module(
|
||||||
target_path=self.target_path,
|
target_path=self.target_path,
|
||||||
|
|||||||
@@ -8,7 +8,7 @@ HELP_MSG_VERSION = "Show the currently installed version of MVT"
|
|||||||
HELP_MSG_OUTPUT = "Specify a path to a folder where you want to store JSON results"
|
HELP_MSG_OUTPUT = "Specify a path to a folder where you want to store JSON results"
|
||||||
HELP_MSG_IOC = "Path to indicators file (can be invoked multiple time)"
|
HELP_MSG_IOC = "Path to indicators file (can be invoked multiple time)"
|
||||||
HELP_MSG_FAST = "Avoid running time/resource consuming features"
|
HELP_MSG_FAST = "Avoid running time/resource consuming features"
|
||||||
HELP_MSG_LIST_MODULES = "Print list of available modules and exit"
|
HELP_MSG_LIST_MODULES = "Print list of available modules and their source, then exit"
|
||||||
HELP_MSG_MODULE = "Name of a single module you would like to run instead of all"
|
HELP_MSG_MODULE = "Name of a single module you would like to run instead of all"
|
||||||
HELP_MSG_LOAD_MODULE = (
|
HELP_MSG_LOAD_MODULE = (
|
||||||
"Load custom MVT module(s) from a Python file or folder "
|
"Load custom MVT module(s) from a Python file or folder "
|
||||||
|
|||||||
@@ -4,18 +4,28 @@
|
|||||||
# https://license.mvt.re/1.1/
|
# https://license.mvt.re/1.1/
|
||||||
|
|
||||||
import hashlib
|
import hashlib
|
||||||
|
import importlib.metadata
|
||||||
import importlib.util
|
import importlib.util
|
||||||
import inspect
|
import inspect
|
||||||
|
import json
|
||||||
import logging
|
import logging
|
||||||
import os
|
import os
|
||||||
import sys
|
import sys
|
||||||
|
from dataclasses import dataclass
|
||||||
|
from functools import lru_cache
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from types import ModuleType
|
from types import ModuleType
|
||||||
from typing import Iterable, Optional
|
from typing import Iterable, Optional
|
||||||
|
|
||||||
from .module import MVTModule
|
from .module import MVTModule
|
||||||
|
from .version import MVT_VERSION
|
||||||
|
|
||||||
MVT_CUSTOM_MODULES_ENV = "MVT_CUSTOM_MODULES"
|
MVT_CUSTOM_MODULES_ENV = "MVT_CUSTOM_MODULES"
|
||||||
|
MODULES_ENTRY_POINT_GROUP = "mvt.modules"
|
||||||
|
EXTERNAL_LOGGER_NAMESPACE = "mvt.ext"
|
||||||
|
PLUGIN_PACKAGE_PREFIX = "mvt_plugin_"
|
||||||
|
_ORIGIN_ATTRIBUTE = "_mvt_module_origin"
|
||||||
|
_PATH_MODULE_PREFIX = "_mvt_custom_module_"
|
||||||
log = logging.getLogger(__name__)
|
log = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
@@ -23,15 +33,74 @@ class CustomModuleLoadError(Exception):
|
|||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class ModuleOrigin:
|
||||||
|
"""Describes where a loaded module came from, for auditability.
|
||||||
|
|
||||||
|
``kind`` is one of ``builtin`` (shipped with MVT), ``package`` (loaded
|
||||||
|
from an installed package) or ``path`` (loaded from a file passed with
|
||||||
|
``--load-module`` or the environment variable).
|
||||||
|
"""
|
||||||
|
|
||||||
|
kind: str
|
||||||
|
name: str
|
||||||
|
version: Optional[str] = None
|
||||||
|
commit: Optional[str] = None
|
||||||
|
file_sha256: Optional[str] = None
|
||||||
|
|
||||||
|
@property
|
||||||
|
def label(self) -> str:
|
||||||
|
label = self.name
|
||||||
|
if self.version:
|
||||||
|
label += f"@{self.version}"
|
||||||
|
label = f"'{label}'"
|
||||||
|
if self.commit:
|
||||||
|
label += f" (commit {self.commit})"
|
||||||
|
if self.file_sha256:
|
||||||
|
label += f" (sha256: {self.file_sha256})"
|
||||||
|
return label
|
||||||
|
|
||||||
|
|
||||||
def _module_name_for_path(path: Path) -> str:
|
def _module_name_for_path(path: Path) -> str:
|
||||||
digest = hashlib.sha256(str(path).encode("utf-8")).hexdigest()[:16]
|
digest = hashlib.sha256(str(path).encode("utf-8")).hexdigest()[:16]
|
||||||
return f"_mvt_custom_module_{path.stem}_{digest}"
|
return f"{_PATH_MODULE_PREFIX}{path.stem}_{digest}"
|
||||||
|
|
||||||
|
|
||||||
|
def get_module_logger(module_class: type[MVTModule]) -> logging.Logger:
|
||||||
|
"""Return the logger a module's records should be emitted through.
|
||||||
|
|
||||||
|
Modules loaded from installed packages or file paths live outside the
|
||||||
|
"mvt" logger hierarchy, so their records would never reach the handlers
|
||||||
|
attached to the "mvt" logger and instead fall through to
|
||||||
|
logging.lastResort (which prints bare messages and drops anything below
|
||||||
|
WARNING). Their loggers are parented under the "mvt.ext" namespace,
|
||||||
|
keeping external module names from colliding with MVT's own logger
|
||||||
|
tree. File-path modules are named after their file instead of the
|
||||||
|
mangled internal import name, and packages following the recommended
|
||||||
|
"mvt_plugin_<name>" naming convention log under "mvt.ext.<name>".
|
||||||
|
"""
|
||||||
|
name = module_class.__module__
|
||||||
|
if name == "mvt" or name.startswith("mvt."):
|
||||||
|
return logging.getLogger(name)
|
||||||
|
|
||||||
|
if name.startswith(_PATH_MODULE_PREFIX):
|
||||||
|
name = Path(get_module_origin(module_class).name).stem
|
||||||
|
else:
|
||||||
|
top_level, separator, rest = name.partition(".")
|
||||||
|
if top_level.startswith(PLUGIN_PACKAGE_PREFIX) and len(top_level) > len(
|
||||||
|
PLUGIN_PACKAGE_PREFIX
|
||||||
|
):
|
||||||
|
name = top_level[len(PLUGIN_PACKAGE_PREFIX) :] + separator + rest
|
||||||
|
|
||||||
|
return logging.getLogger(f"{EXTERNAL_LOGGER_NAMESPACE}.{name}")
|
||||||
|
|
||||||
|
|
||||||
def _iter_module_files(path: Path) -> Iterable[Path]:
|
def _iter_module_files(path: Path) -> Iterable[Path]:
|
||||||
if path.is_file():
|
if path.is_file():
|
||||||
if path.suffix != ".py":
|
if path.suffix != ".py":
|
||||||
raise CustomModuleLoadError(f"Custom module file is not a Python file: {path}")
|
raise CustomModuleLoadError(
|
||||||
|
f"Custom module file is not a Python file: {path}"
|
||||||
|
)
|
||||||
yield path
|
yield path
|
||||||
return
|
return
|
||||||
|
|
||||||
@@ -59,7 +128,9 @@ def _load_python_file(path: Path) -> ModuleType:
|
|||||||
try:
|
try:
|
||||||
spec.loader.exec_module(module)
|
spec.loader.exec_module(module)
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
raise CustomModuleLoadError(f"Unable to import custom module {path}: {exc}") from exc
|
raise CustomModuleLoadError(
|
||||||
|
f"Unable to import custom module {path}: {exc}"
|
||||||
|
) from exc
|
||||||
|
|
||||||
return module
|
return module
|
||||||
|
|
||||||
@@ -84,17 +155,156 @@ def load_custom_modules_from_path(path: str) -> list[type[MVTModule]]:
|
|||||||
resolved_path = Path(path).expanduser().resolve()
|
resolved_path = Path(path).expanduser().resolve()
|
||||||
|
|
||||||
for module_file in _iter_module_files(resolved_path):
|
for module_file in _iter_module_files(resolved_path):
|
||||||
|
file_sha256 = hashlib.sha256(module_file.read_bytes()).hexdigest()
|
||||||
loaded_module = _load_python_file(module_file)
|
loaded_module = _load_python_file(module_file)
|
||||||
|
origin = ModuleOrigin(
|
||||||
|
kind="path", name=str(module_file), file_sha256=file_sha256
|
||||||
|
)
|
||||||
for module_class in discover_mvt_modules(loaded_module):
|
for module_class in discover_mvt_modules(loaded_module):
|
||||||
key = (str(module_file), module_class.__qualname__)
|
key = (str(module_file), module_class.__qualname__)
|
||||||
if key in seen:
|
if key in seen:
|
||||||
continue
|
continue
|
||||||
seen.add(key)
|
seen.add(key)
|
||||||
|
setattr(module_class, _ORIGIN_ATTRIBUTE, origin)
|
||||||
custom_modules.append(module_class)
|
custom_modules.append(module_class)
|
||||||
|
|
||||||
return custom_modules
|
return custom_modules
|
||||||
|
|
||||||
|
|
||||||
|
def _module_key(module_class: type[MVTModule]) -> tuple[str, str]:
|
||||||
|
try:
|
||||||
|
source = str(Path(inspect.getfile(module_class)).resolve())
|
||||||
|
except (OSError, TypeError):
|
||||||
|
source = module_class.__module__
|
||||||
|
return (source, module_class.__qualname__)
|
||||||
|
|
||||||
|
|
||||||
|
def _distribution_commit(dist: importlib.metadata.Distribution) -> Optional[str]:
|
||||||
|
"""Return the VCS commit a distribution was installed from, if recorded.
|
||||||
|
|
||||||
|
Packages installed directly from a repository (``pip install git+...``)
|
||||||
|
record the commit in ``direct_url.json`` (PEP 610).
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
direct_url_text = dist.read_text("direct_url.json")
|
||||||
|
if not direct_url_text:
|
||||||
|
return None
|
||||||
|
commit = json.loads(direct_url_text).get("vcs_info", {}).get("commit_id")
|
||||||
|
return commit if isinstance(commit, str) else None
|
||||||
|
except Exception:
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def _entry_point_origin(entry_point: importlib.metadata.EntryPoint) -> ModuleOrigin:
|
||||||
|
name = entry_point.name
|
||||||
|
version = None
|
||||||
|
commit = None
|
||||||
|
# Manually constructed entry points have no associated distribution.
|
||||||
|
dist = getattr(entry_point, "dist", None)
|
||||||
|
if dist is not None:
|
||||||
|
try:
|
||||||
|
name = dist.name or name
|
||||||
|
version = dist.version
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
commit = _distribution_commit(dist)
|
||||||
|
return ModuleOrigin(kind="package", name=name, version=version, commit=commit)
|
||||||
|
|
||||||
|
|
||||||
|
@lru_cache(maxsize=1)
|
||||||
|
def _packages_distributions() -> dict[str, list[str]]:
|
||||||
|
try:
|
||||||
|
return dict(importlib.metadata.packages_distributions())
|
||||||
|
except Exception:
|
||||||
|
return {}
|
||||||
|
|
||||||
|
|
||||||
|
def get_module_origin(module_class: type[MVTModule]) -> ModuleOrigin:
|
||||||
|
"""Return the origin of a module class for auditing purposes."""
|
||||||
|
origin = module_class.__dict__.get(_ORIGIN_ATTRIBUTE)
|
||||||
|
if isinstance(origin, ModuleOrigin):
|
||||||
|
return origin
|
||||||
|
|
||||||
|
top_level = module_class.__module__.partition(".")[0]
|
||||||
|
if top_level == "mvt":
|
||||||
|
return ModuleOrigin(kind="builtin", name="mvt", version=MVT_VERSION)
|
||||||
|
|
||||||
|
distributions = _packages_distributions().get(top_level)
|
||||||
|
if distributions:
|
||||||
|
name = distributions[0]
|
||||||
|
version = None
|
||||||
|
commit = None
|
||||||
|
try:
|
||||||
|
dist = importlib.metadata.distribution(name)
|
||||||
|
version = dist.version
|
||||||
|
commit = _distribution_commit(dist)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
return ModuleOrigin(kind="package", name=name, version=version, commit=commit)
|
||||||
|
|
||||||
|
try:
|
||||||
|
source = str(Path(inspect.getfile(module_class)).resolve())
|
||||||
|
except (OSError, TypeError):
|
||||||
|
source = module_class.__module__
|
||||||
|
return ModuleOrigin(kind="path", name=source)
|
||||||
|
|
||||||
|
|
||||||
|
def load_installed_modules() -> list[type[MVTModule]]:
|
||||||
|
"""Load MVT modules registered by installed packages.
|
||||||
|
|
||||||
|
Packages register modules in the ``mvt.modules`` entry-point group. Each
|
||||||
|
entry point must resolve to an iterable of MVTModule subclasses, or to a
|
||||||
|
callable which returns one. A broken entry point is skipped with a
|
||||||
|
warning so that a faulty plugin package cannot break MVT.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
entry_points = importlib.metadata.entry_points(group=MODULES_ENTRY_POINT_GROUP)
|
||||||
|
except Exception as exc:
|
||||||
|
log.warning(
|
||||||
|
"Unable to discover installed module packages in entry-point group %s: %s",
|
||||||
|
MODULES_ENTRY_POINT_GROUP,
|
||||||
|
exc,
|
||||||
|
)
|
||||||
|
return []
|
||||||
|
|
||||||
|
installed_modules: list[type[MVTModule]] = []
|
||||||
|
ordered_entry_points = sorted(
|
||||||
|
entry_points, key=lambda entry_point: (entry_point.name, entry_point.value)
|
||||||
|
)
|
||||||
|
for entry_point in ordered_entry_points:
|
||||||
|
try:
|
||||||
|
loaded = entry_point.load()
|
||||||
|
if callable(loaded) and not isinstance(loaded, type):
|
||||||
|
loaded = loaded()
|
||||||
|
module_classes = list(loaded)
|
||||||
|
except (Exception, SystemExit) as exc:
|
||||||
|
log.warning(
|
||||||
|
"Unable to load modules from entry point %s (%s): %s",
|
||||||
|
entry_point.name,
|
||||||
|
entry_point.value,
|
||||||
|
exc,
|
||||||
|
)
|
||||||
|
continue
|
||||||
|
|
||||||
|
origin = _entry_point_origin(entry_point)
|
||||||
|
for module_class in module_classes:
|
||||||
|
if not (
|
||||||
|
isinstance(module_class, type) and issubclass(module_class, MVTModule)
|
||||||
|
):
|
||||||
|
log.warning(
|
||||||
|
"Entry point %s (%s) provided %r which is not an "
|
||||||
|
"MVTModule subclass",
|
||||||
|
entry_point.name,
|
||||||
|
entry_point.value,
|
||||||
|
module_class,
|
||||||
|
)
|
||||||
|
continue
|
||||||
|
setattr(module_class, _ORIGIN_ATTRIBUTE, origin)
|
||||||
|
installed_modules.append(module_class)
|
||||||
|
|
||||||
|
return installed_modules
|
||||||
|
|
||||||
|
|
||||||
def load_custom_modules(paths: Optional[Iterable[str]] = None) -> list[type[MVTModule]]:
|
def load_custom_modules(paths: Optional[Iterable[str]] = None) -> list[type[MVTModule]]:
|
||||||
search_paths: list[str] = []
|
search_paths: list[str] = []
|
||||||
env_path = os.environ.get(MVT_CUSTOM_MODULES_ENV)
|
env_path = os.environ.get(MVT_CUSTOM_MODULES_ENV)
|
||||||
@@ -105,10 +315,17 @@ def load_custom_modules(paths: Optional[Iterable[str]] = None) -> list[type[MVTM
|
|||||||
|
|
||||||
custom_modules: list[type[MVTModule]] = []
|
custom_modules: list[type[MVTModule]] = []
|
||||||
seen: set[tuple[str, str]] = set()
|
seen: set[tuple[str, str]] = set()
|
||||||
|
|
||||||
|
for module_class in load_installed_modules():
|
||||||
|
key = _module_key(module_class)
|
||||||
|
if key in seen:
|
||||||
|
continue
|
||||||
|
seen.add(key)
|
||||||
|
custom_modules.append(module_class)
|
||||||
|
|
||||||
for path in search_paths:
|
for path in search_paths:
|
||||||
for module_class in load_custom_modules_from_path(path):
|
for module_class in load_custom_modules_from_path(path):
|
||||||
source = Path(inspect.getfile(module_class)).resolve()
|
key = _module_key(module_class)
|
||||||
key = (str(source), module_class.__qualname__)
|
|
||||||
if key in seen:
|
if key in seen:
|
||||||
continue
|
continue
|
||||||
seen.add(key)
|
seen.add(key)
|
||||||
|
|||||||
@@ -239,6 +239,13 @@ def init_logging(verbose: bool = False):
|
|||||||
"""
|
"""
|
||||||
log = logging.getLogger("mvt")
|
log = logging.getLogger("mvt")
|
||||||
log.setLevel(logging.DEBUG)
|
log.setLevel(logging.DEBUG)
|
||||||
|
|
||||||
|
# Importing an MVT CLI module calls init_logging() at import time, and
|
||||||
|
# loaded module packages may import one indirectly. Keep this idempotent
|
||||||
|
# so console log lines are not duplicated by a second handler.
|
||||||
|
if any(isinstance(handler, MVTLogHandler) for handler in log.handlers):
|
||||||
|
return
|
||||||
|
|
||||||
consoleHandler = MVTLogHandler()
|
consoleHandler = MVTLogHandler()
|
||||||
consoleHandler.setFormatter(logging.Formatter("%(message)s"))
|
consoleHandler.setFormatter(logging.Formatter("%(message)s"))
|
||||||
if verbose:
|
if verbose:
|
||||||
|
|||||||
@@ -3,10 +3,12 @@ import pytest
|
|||||||
from mvt.common.module import MVTModule
|
from mvt.common.module import MVTModule
|
||||||
from mvt.common.module_loader import (
|
from mvt.common.module_loader import (
|
||||||
CustomModuleLoadError,
|
CustomModuleLoadError,
|
||||||
|
get_module_logger,
|
||||||
load_custom_modules,
|
load_custom_modules,
|
||||||
load_custom_modules_from_path,
|
load_custom_modules_from_path,
|
||||||
module_supports_command,
|
module_supports_command,
|
||||||
)
|
)
|
||||||
|
from mvt.ios.modules.mixed.whatsapp import Whatsapp
|
||||||
|
|
||||||
|
|
||||||
MODULE_TEMPLATE = """
|
MODULE_TEMPLATE = """
|
||||||
@@ -144,3 +146,44 @@ def test_module_supports_command_honors_supported_commands(tmp_path):
|
|||||||
|
|
||||||
assert module_supports_command(module, "ios", "check-backup")
|
assert module_supports_command(module, "ios", "check-backup")
|
||||||
assert not module_supports_command(module, "ios", "check-fs")
|
assert not module_supports_command(module, "ios", "check-fs")
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_module_logger_keeps_builtin_names():
|
||||||
|
assert get_module_logger(Whatsapp).name == "mvt.ios.modules.mixed.whatsapp"
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_module_logger_parents_package_modules_under_mvt_ext():
|
||||||
|
class PackageModule(MVTModule):
|
||||||
|
pass
|
||||||
|
|
||||||
|
PackageModule.__module__ = "some_plugin_package.ios.custom"
|
||||||
|
|
||||||
|
assert (
|
||||||
|
get_module_logger(PackageModule).name
|
||||||
|
== "mvt.ext.some_plugin_package.ios.custom"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_module_logger_strips_the_plugin_package_prefix():
|
||||||
|
class PluginModule(MVTModule):
|
||||||
|
pass
|
||||||
|
|
||||||
|
PluginModule.__module__ = "mvt_plugin_amnesty_custom.ios.custom"
|
||||||
|
|
||||||
|
assert get_module_logger(PluginModule).name == "mvt.ext.amnesty_custom.ios.custom"
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_module_logger_only_strips_the_prefix_from_the_top_level():
|
||||||
|
class NestedModule(MVTModule):
|
||||||
|
pass
|
||||||
|
|
||||||
|
NestedModule.__module__ = "other_package.mvt_plugin_sub"
|
||||||
|
|
||||||
|
assert get_module_logger(NestedModule).name == "mvt.ext.other_package.mvt_plugin_sub"
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_module_logger_names_path_modules_after_their_file(tmp_path):
|
||||||
|
module_path = _write_module(tmp_path / "my_custom_module.py", "PathModule")
|
||||||
|
module = load_custom_modules_from_path(str(module_path))[0]
|
||||||
|
|
||||||
|
assert get_module_logger(module).name == "mvt.ext.my_custom_module"
|
||||||
@@ -8,6 +8,7 @@ import logging
|
|||||||
import os
|
import os
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
|
|
||||||
|
from mvt.common.log import MVTLogHandler
|
||||||
from mvt.common.utils import (
|
from mvt.common.utils import (
|
||||||
CustomJSONEncoder,
|
CustomJSONEncoder,
|
||||||
convert_datetime_to_iso,
|
convert_datetime_to_iso,
|
||||||
@@ -16,6 +17,7 @@ from mvt.common.utils import (
|
|||||||
convert_unix_to_utc_datetime,
|
convert_unix_to_utc_datetime,
|
||||||
generate_hashes_from_path,
|
generate_hashes_from_path,
|
||||||
get_sha256_from_file_path,
|
get_sha256_from_file_path,
|
||||||
|
init_logging,
|
||||||
)
|
)
|
||||||
|
|
||||||
from ..utils import get_artifact_folder
|
from ..utils import get_artifact_folder
|
||||||
@@ -103,3 +105,20 @@ class TestCustomJSONEncoder:
|
|||||||
json.dumps({"name": "家".encode()}, cls=CustomJSONEncoder)
|
json.dumps({"name": "家".encode()}, cls=CustomJSONEncoder)
|
||||||
== '{"name": "\\u5bb6"}'
|
== '{"name": "\\u5bb6"}'
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class TestInitLogging:
|
||||||
|
def test__init_logging_is_idempotent(self):
|
||||||
|
# Loaded module packages may import an MVT CLI module, which calls
|
||||||
|
# init_logging() again at import time. A second call must not add
|
||||||
|
# a duplicate console handler.
|
||||||
|
log = logging.getLogger("mvt")
|
||||||
|
init_logging()
|
||||||
|
handler_count = sum(
|
||||||
|
isinstance(handler, MVTLogHandler) for handler in log.handlers
|
||||||
|
)
|
||||||
|
init_logging()
|
||||||
|
assert (
|
||||||
|
sum(isinstance(handler, MVTLogHandler) for handler in log.handlers)
|
||||||
|
== handler_count
|
||||||
|
)
|
||||||
@@ -1,3 +1,7 @@
|
|||||||
|
import hashlib
|
||||||
|
import importlib.metadata
|
||||||
|
import json
|
||||||
|
|
||||||
from click.testing import CliRunner
|
from click.testing import CliRunner
|
||||||
|
|
||||||
from mvt.android.cli import check_bugreport
|
from mvt.android.cli import check_bugreport
|
||||||
@@ -5,7 +9,9 @@ from mvt.android.cmd_check_androidqf import CmdAndroidCheckAndroidQF
|
|||||||
from mvt.android.cmd_check_backup import CmdAndroidCheckBackup
|
from mvt.android.cmd_check_backup import CmdAndroidCheckBackup
|
||||||
from mvt.android.cmd_check_bugreport import CmdAndroidCheckBugreport
|
from mvt.android.cmd_check_bugreport import CmdAndroidCheckBugreport
|
||||||
from mvt.android.cmd_check_intrusion_logs import CmdAndroidCheckIntrusionLogs
|
from mvt.android.cmd_check_intrusion_logs import CmdAndroidCheckIntrusionLogs
|
||||||
|
from mvt.common import module_loader
|
||||||
from mvt.common.module import MVTModule
|
from mvt.common.module import MVTModule
|
||||||
|
from mvt.common.version import MVT_VERSION
|
||||||
from mvt.ios.cli import check_backup, check_fs
|
from mvt.ios.cli import check_backup, check_fs
|
||||||
|
|
||||||
|
|
||||||
@@ -106,6 +112,150 @@ def test_custom_modules_load_from_environment_without_cli_flag(tmp_path, monkeyp
|
|||||||
assert "EnvBugreportModule" in result.output
|
assert "EnvBugreportModule" in result.output
|
||||||
|
|
||||||
|
|
||||||
|
class InstalledPackageModule(MVTModule):
|
||||||
|
supported_commands = (("ios", "check-backup"),)
|
||||||
|
|
||||||
|
|
||||||
|
def get_installed_package_modules():
|
||||||
|
return [InstalledPackageModule]
|
||||||
|
|
||||||
|
|
||||||
|
def _fake_entry_points(monkeypatch, value, name="test-modules"):
|
||||||
|
entry_point = importlib.metadata.EntryPoint(
|
||||||
|
name=name, value=value, group=module_loader.MODULES_ENTRY_POINT_GROUP
|
||||||
|
)
|
||||||
|
|
||||||
|
def fake_entry_points(*, group):
|
||||||
|
assert group == module_loader.MODULES_ENTRY_POINT_GROUP
|
||||||
|
return [entry_point]
|
||||||
|
|
||||||
|
monkeypatch.setattr(
|
||||||
|
module_loader.importlib.metadata, "entry_points", fake_entry_points
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_installed_module_package_loads_from_entry_point(monkeypatch):
|
||||||
|
_fake_entry_points(monkeypatch, f"{__name__}:get_installed_package_modules")
|
||||||
|
|
||||||
|
modules = module_loader.load_custom_modules()
|
||||||
|
|
||||||
|
assert modules == [InstalledPackageModule]
|
||||||
|
|
||||||
|
|
||||||
|
def test_broken_module_entry_point_is_skipped(monkeypatch, caplog):
|
||||||
|
_fake_entry_points(monkeypatch, "nonexistent_module_xyz:get_modules")
|
||||||
|
|
||||||
|
with caplog.at_level("WARNING"):
|
||||||
|
modules = module_loader.load_custom_modules()
|
||||||
|
|
||||||
|
assert modules == []
|
||||||
|
assert "Unable to load modules from entry point" in caplog.text
|
||||||
|
|
||||||
|
|
||||||
|
def test_entry_point_module_deduplicated_against_paths(monkeypatch, tmp_path):
|
||||||
|
_fake_entry_points(monkeypatch, f"{__name__}:get_installed_package_modules")
|
||||||
|
module_path = _write_custom_module(
|
||||||
|
tmp_path / "custom.py",
|
||||||
|
"PathLoadedModule",
|
||||||
|
(("ios", "check-backup"),),
|
||||||
|
)
|
||||||
|
|
||||||
|
modules = module_loader.load_custom_modules([str(module_path)])
|
||||||
|
|
||||||
|
assert [module.__name__ for module in modules] == [
|
||||||
|
"InstalledPackageModule",
|
||||||
|
"PathLoadedModule",
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
def test_list_modules_shows_module_sources(tmp_path, caplog):
|
||||||
|
module_path = _write_custom_module(
|
||||||
|
tmp_path / "custom.py",
|
||||||
|
"SourcedBackupModule",
|
||||||
|
(("ios", "check-backup"),),
|
||||||
|
)
|
||||||
|
file_sha256 = hashlib.sha256(module_path.read_bytes()).hexdigest()
|
||||||
|
custom_modules = module_loader.load_custom_modules([str(module_path)])
|
||||||
|
|
||||||
|
from mvt.ios.cmd_check_backup import CmdIOSCheckBackup
|
||||||
|
|
||||||
|
cmd = CmdIOSCheckBackup(target_path=str(tmp_path), custom_modules=custom_modules)
|
||||||
|
cmd.list_modules()
|
||||||
|
|
||||||
|
assert f" - Modules from 'mvt@{MVT_VERSION}':" in caplog.text
|
||||||
|
assert (
|
||||||
|
f" - Modules from '{module_path}' (sha256: {file_sha256}): SourcedBackupModule"
|
||||||
|
in caplog.text
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_builtin_module_origin():
|
||||||
|
from mvt.ios.modules.backup import BACKUP_MODULES
|
||||||
|
|
||||||
|
origin = module_loader.get_module_origin(BACKUP_MODULES[0])
|
||||||
|
|
||||||
|
assert origin.kind == "builtin"
|
||||||
|
assert origin.name == "mvt"
|
||||||
|
assert origin.version == MVT_VERSION
|
||||||
|
|
||||||
|
|
||||||
|
def test_installed_module_origin(monkeypatch):
|
||||||
|
_fake_entry_points(monkeypatch, f"{__name__}:get_installed_package_modules")
|
||||||
|
|
||||||
|
modules = module_loader.load_custom_modules()
|
||||||
|
|
||||||
|
origin = module_loader.get_module_origin(modules[0])
|
||||||
|
assert origin.kind == "package"
|
||||||
|
assert origin.name == "test-modules"
|
||||||
|
|
||||||
|
|
||||||
|
def test_distribution_commit_read_from_direct_url():
|
||||||
|
class FakeDistribution:
|
||||||
|
def read_text(self, filename):
|
||||||
|
assert filename == "direct_url.json"
|
||||||
|
return json.dumps(
|
||||||
|
{
|
||||||
|
"url": "https://github.com/example/example-modules",
|
||||||
|
"vcs_info": {"commit_id": "abc1234", "vcs": "git"},
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
assert module_loader._distribution_commit(FakeDistribution()) == "abc1234"
|
||||||
|
|
||||||
|
|
||||||
|
def test_command_log_records_loaded_modules(tmp_path):
|
||||||
|
(tmp_path / "Manifest.db").touch()
|
||||||
|
(tmp_path / "Info.plist").touch()
|
||||||
|
module_path = _write_custom_module(
|
||||||
|
tmp_path / "custom.py",
|
||||||
|
"AuditedRunModule",
|
||||||
|
(("ios", "check-backup"),),
|
||||||
|
slug="audited_run_module",
|
||||||
|
)
|
||||||
|
file_sha256 = hashlib.sha256(module_path.read_bytes()).hexdigest()
|
||||||
|
output_path = tmp_path / "out"
|
||||||
|
|
||||||
|
result = CliRunner().invoke(
|
||||||
|
check_backup,
|
||||||
|
[
|
||||||
|
"--module",
|
||||||
|
"AuditedRunModule",
|
||||||
|
"--load-module",
|
||||||
|
str(module_path),
|
||||||
|
"--output",
|
||||||
|
str(output_path),
|
||||||
|
str(tmp_path),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
|
||||||
|
assert result.exit_code == 0
|
||||||
|
command_log = (output_path / "command.log").read_text(encoding="utf-8")
|
||||||
|
assert (
|
||||||
|
f"Loaded 1 check-backup modules from '{module_path}' "
|
||||||
|
f"(sha256: {file_sha256}): AuditedRunModule" in command_log
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
class NestedBugreportModule(MVTModule):
|
class NestedBugreportModule(MVTModule):
|
||||||
supported_commands = (("android", "check-bugreport"),)
|
supported_commands = (("android", "check-bugreport"),)
|
||||||
|
|
||||||
|
|||||||
Reference in new issue
Block a user