From 30e25f9692c1888764f095195c48f2bce7acd3c3 Mon Sep 17 00:00:00 2001 From: Janik Besendorf Date: Sun, 23 Aug 2026 10:06:01 +0200 Subject: [PATCH] feat: add versioned output schemas --- docs/development.md | 4 + docs/output-schemas.md | 77 ++++++++++ mkdocs.yml | 1 + .../android/artifacts/tombstone_crashes.py | 9 +- src/mvt/android/cli.py | 17 ++- src/mvt/android/modules/androidqf/__init__.py | 3 +- .../android/modules/androidqf/aqf_settings.py | 3 + src/mvt/android/modules/androidqf/base.py | 5 +- src/mvt/android/modules/backup/__init__.py | 3 +- src/mvt/android/modules/backup/base.py | 5 +- src/mvt/android/modules/bugreport/__init__.py | 3 +- src/mvt/android/modules/bugreport/base.py | 5 +- .../modules/bugreport/dumpsys_receivers.py | 3 + .../modules/intrusion_logs/__init__.py | 3 +- .../android/modules/intrusion_logs/base.py | 5 +- src/mvt/common/cmd_schemas.py | 22 +++ src/mvt/common/command.py | 14 +- src/mvt/common/module.py | 52 +++---- src/mvt/ios/cli.py | 17 ++- src/mvt/ios/modules/backup/__init__.py | 8 +- src/mvt/ios/modules/backup/backup_info.py | 3 + src/mvt/ios/modules/base.py | 5 +- src/mvt/ios/modules/fs/__init__.py | 3 +- src/mvt/ios/modules/fs/cache_files.py | 3 + src/mvt/ios/modules/mixed/__init__.py | 3 +- .../mixed/webkit_session_resource_log.py | 3 + src/mvt/schemas/__init__.py | 49 +++++++ src/mvt/schemas/models.py | 101 +++++++++++++ src/mvt/schemas/registry.py | 133 ++++++++++++++++++ src/mvt/schemas/serialization.py | 35 +++++ tests/common/test_schemas.py | 131 +++++++++++++++++ 31 files changed, 688 insertions(+), 40 deletions(-) create mode 100644 docs/output-schemas.md create mode 100644 src/mvt/common/cmd_schemas.py create mode 100644 src/mvt/schemas/__init__.py create mode 100644 src/mvt/schemas/models.py create mode 100644 src/mvt/schemas/registry.py create mode 100644 src/mvt/schemas/serialization.py create mode 100644 tests/common/test_schemas.py diff --git a/docs/development.md b/docs/development.md index 53a8f61..f8a501d 100644 --- a/docs/development.md +++ b/docs/development.md @@ -82,6 +82,10 @@ class ExampleCustomModule(MVTModule): return None ``` +Custom modules should declare an `output_model` using a Pydantic `RootModel`. +This validates saved results and lets consumers obtain the module's JSON Schema. +See [Output schemas](output-schemas.md#custom-modules) for a complete example. + 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 diff --git a/docs/output-schemas.md b/docs/output-schemas.md new file mode 100644 index 0000000..bbb2a32 --- /dev/null +++ b/docs/output-schemas.md @@ -0,0 +1,77 @@ +# Output schemas + +MVT validates its JSON output with versioned Pydantic models. The schema version +used for a run is recorded as `output_schema_version` in `info.json`. + +MVT preserves the established on-disk format: most module files contain an array +of result objects, while modules whose results are naturally grouped contain an +object keyed by source or namespace. Detection files and `alerts.json` contain +arrays of alert objects. Timestamps remain strings because their timezone and +precision depend on the source artifact. + +## Exporting JSON Schema + +Both platform commands can print a versioned JSON Schema bundle: + +```bash +mvt-ios schemas +mvt-android schemas +``` + +Use `--output` to write one Draft 2020-12 JSON Schema file for each output: + +```bash +mvt-ios schemas --output ./mvt-ios-schemas +mvt-android schemas --output ./mvt-android-schemas +``` + +## Python API + +Models and schema discovery functions are available from `mvt.schemas`: + +```python +from mvt.schemas import get_output_model + +SafariHistoryOutput = get_output_model("safari_history", platform="ios") +validated = SafariHistoryOutput.model_validate(records) +json_schema = SafariHistoryOutput.model_json_schema() +``` + +Common outputs have dedicated field-level models. Built-in module outputs have a +declared root shape, and modules with an established dedicated record model expose +its complete field schema. + +## Custom modules + +Custom modules can publish a precise contract by assigning a Pydantic root model +to `output_model`: + +```python +from pydantic import BaseModel, RootModel + +from mvt.common.module import MVTModule + + +class ExampleRecord(BaseModel): + message: str + timestamp: str | None = None + + +class ExampleOutput(RootModel[list[ExampleRecord]]): + pass + + +class ExampleModule(MVTModule): + output_model = ExampleOutput +``` + +For compatibility, a custom module without `output_model` can still write an +array or object containing JSON values. MVT logs a warning when it uses this +generic contract. A future major release may require custom modules to declare +their output models. + +## Compatibility policy + +Within one output schema major version, required fields are not removed and field +types are not narrowed. New optional fields may be added. A breaking output +change requires a new schema major version and a migration note. diff --git a/mkdocs.yml b/mkdocs.yml index a7b22ac..044e1a7 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -31,6 +31,7 @@ nav: - Introduction: "introduction.md" - Installation: "install.md" - Command Completion: "command_completion.md" + - Output Schemas: "output-schemas.md" - Using Docker: "docker.md" - MVT for iOS: - iOS Forensic Methodology: "ios/methodology.md" diff --git a/src/mvt/android/artifacts/tombstone_crashes.py b/src/mvt/android/artifacts/tombstone_crashes.py index bcfb389..d31ebd1 100644 --- a/src/mvt/android/artifacts/tombstone_crashes.py +++ b/src/mvt/android/artifacts/tombstone_crashes.py @@ -4,7 +4,7 @@ # https://license.mvt.re/1.1/ import datetime -from typing import List, Optional +from typing import ClassVar, List, Optional import pydantic import betterproto2 @@ -13,6 +13,7 @@ from dateutil import parser from mvt.android.parsers.proto.tombstone import Tombstone from mvt.common.module_types import ModuleAtomicResult, ModuleSerializedResult from mvt.common.utils import convert_datetime_to_iso +from mvt.schemas import OutputModel from .artifact import AndroidArtifact @@ -73,6 +74,10 @@ class TombstoneCrashResult(pydantic.BaseModel): extra: Optional[str] = None +class TombstoneCrashOutput(pydantic.RootModel[List[TombstoneCrashResult]]): + """Complete JSON document written by tombstone crash modules.""" + + class TombstoneCrashArtifact(AndroidArtifact): """ Parser for Android tombstone crash files. @@ -80,6 +85,8 @@ class TombstoneCrashArtifact(AndroidArtifact): This parser can parse both text and protobuf tombstone crash files. """ + output_model: ClassVar[OutputModel] = TombstoneCrashOutput + def serialize(self, record: ModuleAtomicResult) -> ModuleSerializedResult: return { "timestamp": record["timestamp"], diff --git a/src/mvt/android/cli.py b/src/mvt/android/cli.py index ca7e3dc..7cfc50b 100644 --- a/src/mvt/android/cli.py +++ b/src/mvt/android/cli.py @@ -15,6 +15,7 @@ from mvt.common.cli_plugins import ( register_cli_plugins, ) from mvt.common.cmd_check_iocs import CmdCheckIOCS +from mvt.common.cmd_schemas import emit_schemas from mvt.common.completion import ( SUPPORTED_SHELLS, completion_instructions, @@ -102,7 +103,7 @@ def cli(ctx, disable_update_check, disable_indicator_update_check): ctx.ensure_object(dict) ctx.obj["disable_version_check"] = disable_update_check ctx.obj["disable_indicator_check"] = disable_indicator_update_check - if ctx.invoked_subcommand != "completion": + if ctx.invoked_subcommand not in ("completion", "schemas"): logo( disable_version_check=disable_update_check, disable_indicator_check=disable_indicator_update_check, @@ -117,6 +118,20 @@ def version(): return +# ============================================================================== +# Command: schemas +# ============================================================================== +@cli.command("schemas", help="Print or export the JSON Schemas for MVT outputs.") +@click.option( + "--output", + "-o", + type=click.Path(file_okay=False, dir_okay=True), + help="Write one JSON Schema file per output to this directory.", +) +def schemas(output): + emit_schemas("android", output) + + # ============================================================================== # Command: completion # ============================================================================== diff --git a/src/mvt/android/modules/androidqf/__init__.py b/src/mvt/android/modules/androidqf/__init__.py index 1d8d619..39dc3fa 100644 --- a/src/mvt/android/modules/androidqf/__init__.py +++ b/src/mvt/android/modules/androidqf/__init__.py @@ -11,8 +11,9 @@ from .aqf_processes import AQFProcesses from .aqf_settings import AQFSettings from .mounts import Mounts from .root_binaries import RootBinaries +from mvt.common.module import MVTModule -ANDROIDQF_MODULES = [ +ANDROIDQF_MODULES: list[type[MVTModule]] = [ AQFPackages, AQFProcesses, AQFGetProp, diff --git a/src/mvt/android/modules/androidqf/aqf_settings.py b/src/mvt/android/modules/androidqf/aqf_settings.py index 8d5bb51..e7d39b8 100644 --- a/src/mvt/android/modules/androidqf/aqf_settings.py +++ b/src/mvt/android/modules/androidqf/aqf_settings.py @@ -8,6 +8,7 @@ from typing import Optional from mvt.android.artifacts.settings import Settings as SettingsArtifact from mvt.common.module_types import ModuleResults +from mvt.schemas import MappingOutput from .base import AndroidQFModule @@ -15,6 +16,8 @@ from .base import AndroidQFModule class AQFSettings(SettingsArtifact, AndroidQFModule): """This module analyse setting files""" + output_model = MappingOutput + def __init__( self, file_path: Optional[str] = None, diff --git a/src/mvt/android/modules/androidqf/base.py b/src/mvt/android/modules/androidqf/base.py index b0304d0..64f55ef 100644 --- a/src/mvt/android/modules/androidqf/base.py +++ b/src/mvt/android/modules/androidqf/base.py @@ -7,15 +7,18 @@ import fnmatch import logging import os import zipfile -from typing import List, Optional +from typing import ClassVar, List, Optional from mvt.common.module import MVTModule from mvt.common.module_types import ModuleResults +from mvt.schemas import OutputModel, RecordListOutput class AndroidQFModule(MVTModule): """This class provides a base for all Android Data analysis modules.""" + output_model: ClassVar[OutputModel] = RecordListOutput + def __init__( self, file_path: Optional[str] = None, diff --git a/src/mvt/android/modules/backup/__init__.py b/src/mvt/android/modules/backup/__init__.py index 47a5e1c..1806349 100644 --- a/src/mvt/android/modules/backup/__init__.py +++ b/src/mvt/android/modules/backup/__init__.py @@ -4,5 +4,6 @@ # https://license.mvt.re/1.1/ from .sms import SMS +from mvt.common.module import MVTModule -BACKUP_MODULES = [SMS] +BACKUP_MODULES: list[type[MVTModule]] = [SMS] diff --git a/src/mvt/android/modules/backup/base.py b/src/mvt/android/modules/backup/base.py index 6383e4b..dbbb13f 100644 --- a/src/mvt/android/modules/backup/base.py +++ b/src/mvt/android/modules/backup/base.py @@ -7,14 +7,17 @@ import fnmatch import logging import os from tarfile import TarFile -from typing import List, Optional +from typing import ClassVar, List, Optional from mvt.common.module import ModuleResults, MVTModule +from mvt.schemas import OutputModel, RecordListOutput class BackupModule(MVTModule): """This class provides a base for all backup extractios modules""" + output_model: ClassVar[OutputModel] = RecordListOutput + def __init__( self, file_path: Optional[str] = None, diff --git a/src/mvt/android/modules/bugreport/__init__.py b/src/mvt/android/modules/bugreport/__init__.py index 1594af9..17dd226 100644 --- a/src/mvt/android/modules/bugreport/__init__.py +++ b/src/mvt/android/modules/bugreport/__init__.py @@ -16,8 +16,9 @@ from .dumpsys_receivers import DumpsysReceivers from .dumpsys_adb_state import DumpsysADBState from .fs_timestamps import BugReportTimestamps from .tombstones import Tombstones +from mvt.common.module import MVTModule -BUGREPORT_MODULES = [ +BUGREPORT_MODULES: list[type[MVTModule]] = [ DumpsysAccessibility, DumpsysActivities, DumpsysAppops, diff --git a/src/mvt/android/modules/bugreport/base.py b/src/mvt/android/modules/bugreport/base.py index 156e01c..693dbcd 100644 --- a/src/mvt/android/modules/bugreport/base.py +++ b/src/mvt/android/modules/bugreport/base.py @@ -7,15 +7,18 @@ import fnmatch import logging import os from pathlib import Path -from typing import List, Optional +from typing import ClassVar, List, Optional from zipfile import ZipFile from mvt.common.module import ModuleResults, MVTModule +from mvt.schemas import OutputModel, RecordListOutput class BugReportModule(MVTModule): """This class provides a base for all Android Bug Report modules.""" + output_model: ClassVar[OutputModel] = RecordListOutput + def __init__( self, file_path: Optional[str] = None, diff --git a/src/mvt/android/modules/bugreport/dumpsys_receivers.py b/src/mvt/android/modules/bugreport/dumpsys_receivers.py index 8907d5e..b3abb5b 100644 --- a/src/mvt/android/modules/bugreport/dumpsys_receivers.py +++ b/src/mvt/android/modules/bugreport/dumpsys_receivers.py @@ -8,6 +8,7 @@ from typing import Optional from mvt.android.artifacts.dumpsys_receivers import DumpsysReceiversArtifact from mvt.common.module_types import ModuleResults +from mvt.schemas import MappingOutput from .base import BugReportModule @@ -15,6 +16,8 @@ from .base import BugReportModule class DumpsysReceivers(DumpsysReceiversArtifact, BugReportModule): """This module extracts details on receivers for risky activities.""" + output_model = MappingOutput + def __init__( self, file_path: Optional[str] = None, diff --git a/src/mvt/android/modules/intrusion_logs/__init__.py b/src/mvt/android/modules/intrusion_logs/__init__.py index f8be973..2f2a118 100644 --- a/src/mvt/android/modules/intrusion_logs/__init__.py +++ b/src/mvt/android/modules/intrusion_logs/__init__.py @@ -6,8 +6,9 @@ from .connect_event import ConnectEvent from .dns_event import DnsEvent from .security_event import SecurityEvent +from mvt.common.module import MVTModule -INTRUSION_LOGS_MODULES = [ +INTRUSION_LOGS_MODULES: list[type[MVTModule]] = [ DnsEvent, ConnectEvent, SecurityEvent, diff --git a/src/mvt/android/modules/intrusion_logs/base.py b/src/mvt/android/modules/intrusion_logs/base.py index 4aac618..fb8fb13 100644 --- a/src/mvt/android/modules/intrusion_logs/base.py +++ b/src/mvt/android/modules/intrusion_logs/base.py @@ -9,7 +9,7 @@ import json import logging import zipfile from pathlib import Path -from typing import Optional, Union +from typing import ClassVar, Optional, Union try: import zoneinfo @@ -18,6 +18,7 @@ except ImportError: from mvt.common.module import MVTModule from mvt.common.utils import convert_datetime_to_iso, convert_unix_to_iso +from mvt.schemas import OutputModel, RecordListOutput class IntrusionLogsModule(MVTModule): @@ -41,6 +42,8 @@ class IntrusionLogsModule(MVTModule): original file-loading code path. """ + output_model: ClassVar[OutputModel] = RecordListOutput + def __init__( self, file_path: Optional[str] = None, diff --git a/src/mvt/common/cmd_schemas.py b/src/mvt/common/cmd_schemas.py new file mode 100644 index 0000000..bea1778 --- /dev/null +++ b/src/mvt/common/cmd_schemas.py @@ -0,0 +1,22 @@ +# Mobile Verification Toolkit (MVT) +# Copyright (c) 2026 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/ + +"""Shared implementation for the platform schema-export commands.""" + +import json +from typing import Optional + +import click + +from mvt.schemas import export_json_schemas, schema_bundle + + +def emit_schemas(platform: str, output: Optional[str]) -> None: + if output: + paths = export_json_schemas(output, platform) + click.echo(f"Exported {len(paths)} {platform} output schemas to {output}") + return + + click.echo(json.dumps(schema_bundle(platform), indent=2)) diff --git a/src/mvt/common/command.py b/src/mvt/common/command.py index 6111dc7..ab2f584 100644 --- a/src/mvt/common/command.py +++ b/src/mvt/common/command.py @@ -33,6 +33,13 @@ from .utils import ( get_sha256_from_file_path, ) from .version import MVT_VERSION +from mvt.schemas import ( + OUTPUT_SCHEMA_VERSION, + AlertResults, + RunInfo, + URLResults, + validate_output, +) class Command: @@ -152,6 +159,8 @@ class Command: if not alerts: return + alerts = validate_output(AlertResults, alerts) + alerts_path = os.path.join(self.results_path, "alerts.json") with open(alerts_path, "w+", encoding="utf-8") as handle: json.dump(alerts, handle, indent=4, cls=CustomJSONEncoder) @@ -160,9 +169,10 @@ class Command: if not self.results_path or not self.url_results: return + urls = validate_output(URLResults, self.url_results) urls_path = os.path.join(self.results_path, "urls.json") with open(urls_path, "w", encoding="utf-8") as handle: - json.dump(self.url_results, handle, indent=4, cls=CustomJSONEncoder) + json.dump(urls, handle, indent=4, cls=CustomJSONEncoder) def _store_alerts_timeline(self) -> None: if not self.results_path: @@ -185,6 +195,7 @@ class Command: "date": convert_datetime_to_iso(datetime.now()), "ioc_files": [], "hashes": [], + "output_schema_version": OUTPUT_SCHEMA_VERSION, } for coll in self.iocs.ioc_collections: @@ -197,6 +208,7 @@ class Command: info["hashes"] = self.hash_values + info = validate_output(RunInfo, info) info_path = os.path.join(self.results_path, "info.json") with open(info_path, "w+", encoding="utf-8") as handle: json.dump(info, handle, indent=4) diff --git a/src/mvt/common/module.py b/src/mvt/common/module.py index cd127a6..13439a6 100644 --- a/src/mvt/common/module.py +++ b/src/mvt/common/module.py @@ -9,7 +9,12 @@ import logging import os import re from dataclasses import asdict, is_dataclass -from typing import Any, Dict, Optional, Sequence +from typing import Any, ClassVar, Dict, Optional, Sequence + +from pydantic import BaseModel, ValidationError + +from mvt.schemas.models import AlertResults, GenericModuleOutput, TimelineResults +from mvt.schemas.serialization import validate_output from .alerts import AlertStore from .indicators import Indicators @@ -46,6 +51,7 @@ class MVTModule: slug: Optional[str] = None dependencies: Sequence[type["MVTModule"]] = () supported_commands: Sequence[tuple[str, str]] = () + output_model: ClassVar[type[BaseModel]] = GenericModuleOutput def __init__( self, @@ -95,7 +101,7 @@ class MVTModule: @classmethod def from_json(cls, json_path: str, log: logging.Logger): with open(json_path, "r", encoding="utf-8") as handle: - results = json.load(handle) + results = validate_output(cls.output_model, json.load(handle)) if log: log.info('Loaded %d results from "%s"', len(results), json_path) @@ -136,38 +142,35 @@ class MVTModule: name = self.get_slug() if self.results: - converted_results: Any - if isinstance(self.results, dict): - converted_results = self.results - else: - converted_results = [ - asdict(result) - if is_dataclass(result) and not isinstance(result, type) - else result - for result in self.results - ] results_file_name = f"{name}.json" results_json_path = os.path.join(self.results_path, results_file_name) - with open(results_json_path, "w", encoding="utf-8") as handle: - try: + if self.output_model is GenericModuleOutput: + self.log.warning( + "Module %s does not declare an output_model; using the " + "generic compatibility schema", + self.__class__.__name__, + ) + try: + converted_results = validate_output(self.output_model, self.results) + except (TypeError, ValueError, ValidationError) as exc: + self.log.error( + "Output from module %s does not match schema %s: %s", + self.__class__.__name__, + self.output_model.__name__, + exc, + ) + else: + with open(results_json_path, "w", encoding="utf-8") as handle: json.dump( converted_results, handle, indent=4, cls=CustomJSONEncoder ) - except Exception as exc: - self.log.error( - "Unable to store results of module %s to file %s: %s", - self.__class__.__name__, - results_file_name, - exc, - ) if self.alertstore.alerts: detected_file_name = f"{name}_detected.json" detected_json_path = os.path.join(self.results_path, detected_file_name) + alerts = validate_output(AlertResults, self.alertstore.as_json()) with open(detected_json_path, "w", encoding="utf-8") as handle: - json.dump( - self.alertstore.as_json(), handle, indent=4, cls=CustomJSONEncoder - ) + json.dump(alerts, handle, indent=4, cls=CustomJSONEncoder) def serialize(self, result: ModuleAtomicResult) -> ModuleSerializedResult: raise NotImplementedError @@ -298,6 +301,7 @@ def save_timeline(timeline: list, timeline_path: str, is_utc: bool = True) -> No :param timeline_path: Path to the csv file to store the timeline to """ + timeline = validate_output(TimelineResults, timeline) with open(timeline_path, "w", encoding="utf-8") as handle: csvoutput = csv.writer( handle, delimiter=",", quotechar='"', quoting=csv.QUOTE_ALL, escapechar="\\" diff --git a/src/mvt/ios/cli.py b/src/mvt/ios/cli.py index c338fa5..ad7c36c 100644 --- a/src/mvt/ios/cli.py +++ b/src/mvt/ios/cli.py @@ -15,6 +15,7 @@ from mvt.common.cli_plugins import ( register_cli_plugins, ) from mvt.common.cmd_check_iocs import CmdCheckIOCS +from mvt.common.cmd_schemas import emit_schemas from mvt.common.completion import ( SUPPORTED_SHELLS, completion_instructions, @@ -106,7 +107,7 @@ def cli(ctx, disable_update_check, disable_indicator_update_check): ctx.ensure_object(dict) ctx.obj["disable_version_check"] = disable_update_check ctx.obj["disable_indicator_check"] = disable_indicator_update_check - if ctx.invoked_subcommand != "completion": + if ctx.invoked_subcommand not in ("completion", "schemas"): logo( disable_version_check=disable_update_check, disable_indicator_check=disable_indicator_update_check, @@ -121,6 +122,20 @@ def version(): return +# ============================================================================== +# Command: schemas +# ============================================================================== +@cli.command("schemas", help="Print or export the JSON Schemas for MVT outputs.") +@click.option( + "--output", + "-o", + type=click.Path(file_okay=False, dir_okay=True), + help="Write one JSON Schema file per output to this directory.", +) +def schemas(output): + emit_schemas("ios", output) + + # ============================================================================== # Command: completion # ============================================================================== diff --git a/src/mvt/ios/modules/backup/__init__.py b/src/mvt/ios/modules/backup/__init__.py index 0fb833a..033ab63 100644 --- a/src/mvt/ios/modules/backup/__init__.py +++ b/src/mvt/ios/modules/backup/__init__.py @@ -7,5 +7,11 @@ from .backup_info import BackupInfo from .configuration_profiles import ConfigurationProfiles from .manifest import Manifest from .profile_events import ProfileEvents +from mvt.common.module import MVTModule -BACKUP_MODULES = [BackupInfo, ConfigurationProfiles, Manifest, ProfileEvents] +BACKUP_MODULES: list[type[MVTModule]] = [ + BackupInfo, + ConfigurationProfiles, + Manifest, + ProfileEvents, +] diff --git a/src/mvt/ios/modules/backup/backup_info.py b/src/mvt/ios/modules/backup/backup_info.py index 6aadd45..7f4883e 100644 --- a/src/mvt/ios/modules/backup/backup_info.py +++ b/src/mvt/ios/modules/backup/backup_info.py @@ -10,6 +10,7 @@ from typing import Optional from mvt.common.module import DatabaseNotFoundError from mvt.common.module_types import ModuleResults +from mvt.schemas import MappingOutput from mvt.ios.versions import get_device_desc_from_id, is_ios_version_outdated from ..base import IOSExtraction @@ -18,6 +19,8 @@ from ..base import IOSExtraction class BackupInfo(IOSExtraction): """This module extracts information about the device and the backup.""" + output_model = MappingOutput + def __init__( self, file_path: Optional[str] = None, diff --git a/src/mvt/ios/modules/base.py b/src/mvt/ios/modules/base.py index a5c3354..6f79a21 100644 --- a/src/mvt/ios/modules/base.py +++ b/src/mvt/ios/modules/base.py @@ -11,7 +11,7 @@ import sqlite3 import subprocess import tempfile from pathlib import Path -from typing import Iterator, Optional, Union +from typing import ClassVar, Iterator, Optional, Union from mvt.common.module import ( DatabaseCorruptedError, @@ -19,6 +19,7 @@ from mvt.common.module import ( ModuleResults, MVTModule, ) +from mvt.schemas import OutputModel, RecordListOutput class TemporarySQLiteConnection(sqlite3.Connection): @@ -39,6 +40,8 @@ class IOSExtraction(MVTModule): """This class provides a base for all iOS filesystem/backup extraction modules.""" + output_model: ClassVar[OutputModel] = RecordListOutput + def __init__( self, file_path: Optional[str] = None, diff --git a/src/mvt/ios/modules/fs/__init__.py b/src/mvt/ios/modules/fs/__init__.py index 06dba24..5ffb6da 100644 --- a/src/mvt/ios/modules/fs/__init__.py +++ b/src/mvt/ios/modules/fs/__init__.py @@ -14,8 +14,9 @@ from .version_history import IOSVersionHistory from .webkit_indexeddb import WebkitIndexedDB from .webkit_localstorage import WebkitLocalStorage from .webkit_safariviewservice import WebkitSafariViewService +from mvt.common.module import MVTModule -FS_MODULES = [ +FS_MODULES: list[type[MVTModule]] = [ CacheFiles, Filesystem, Netusage, diff --git a/src/mvt/ios/modules/fs/cache_files.py b/src/mvt/ios/modules/fs/cache_files.py index 54a34eb..385fbab 100644 --- a/src/mvt/ios/modules/fs/cache_files.py +++ b/src/mvt/ios/modules/fs/cache_files.py @@ -13,11 +13,14 @@ from mvt.common.module_types import ( ModuleResults, ModuleSerializedResult, ) +from mvt.schemas import MappingOutput from ..base import IOSExtraction class CacheFiles(IOSExtraction): + output_model = MappingOutput + def __init__( self, file_path: Optional[str] = None, diff --git a/src/mvt/ios/modules/mixed/__init__.py b/src/mvt/ios/modules/mixed/__init__.py index a846bf2..ab554bb 100644 --- a/src/mvt/ios/modules/mixed/__init__.py +++ b/src/mvt/ios/modules/mixed/__init__.py @@ -27,8 +27,9 @@ from .webkit_resource_load_statistics import WebkitResourceLoadStatistics from .webkit_session_resource_log import WebkitSessionResourceLog from .whatsapp import Whatsapp from .whatsapp_contacts import WhatsappContacts +from mvt.common.module import MVTModule -MIXED_MODULES = [ +MIXED_MODULES: list[type[MVTModule]] = [ Calls, ChromeFavicon, ChromeHistory, diff --git a/src/mvt/ios/modules/mixed/webkit_session_resource_log.py b/src/mvt/ios/modules/mixed/webkit_session_resource_log.py index eb34311..753f9bd 100644 --- a/src/mvt/ios/modules/mixed/webkit_session_resource_log.py +++ b/src/mvt/ios/modules/mixed/webkit_session_resource_log.py @@ -10,6 +10,7 @@ from typing import Optional from mvt.common.module_types import ModuleResults from mvt.common.utils import convert_datetime_to_iso +from mvt.schemas import MappingOutput from ..base import IOSExtraction @@ -32,6 +33,8 @@ class WebkitSessionResourceLog(IOSExtraction): """ + output_model = MappingOutput + def __init__( self, file_path: Optional[str] = None, diff --git a/src/mvt/schemas/__init__.py b/src/mvt/schemas/__init__.py new file mode 100644 index 0000000..eaa90aa --- /dev/null +++ b/src/mvt/schemas/__init__.py @@ -0,0 +1,49 @@ +"""Stable public API for MVT output models and JSON Schemas.""" + +from .models import ( + OUTPUT_SCHEMA_VERSION, + AlertResult, + AlertResults, + FileHash, + GenericModuleOutput, + MappingOutput, + ModuleRecord, + OutputModel, + RecordListOutput, + RunInfo, + TimelineEvent, + TimelineResults, + URLResult, + URLResults, +) +from .registry import ( + export_json_schemas, + get_output_model, + get_output_models, + schema_bundle, +) +from .serialization import json_compatible, validate_output, write_output + +__all__ = [ + "OUTPUT_SCHEMA_VERSION", + "AlertResult", + "AlertResults", + "FileHash", + "GenericModuleOutput", + "MappingOutput", + "ModuleRecord", + "OutputModel", + "RecordListOutput", + "RunInfo", + "TimelineEvent", + "TimelineResults", + "URLResult", + "URLResults", + "export_json_schemas", + "get_output_model", + "get_output_models", + "json_compatible", + "schema_bundle", + "validate_output", + "write_output", +] diff --git a/src/mvt/schemas/models.py b/src/mvt/schemas/models.py new file mode 100644 index 0000000..0f4a204 --- /dev/null +++ b/src/mvt/schemas/models.py @@ -0,0 +1,101 @@ +# Mobile Verification Toolkit (MVT) +# Copyright (c) 2026 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/ + +"""Public Pydantic models for files written by MVT. + +Module records intentionally remain extensible: each extraction module has its own +record fields, while the root models make the top-level output shape predictable. +More specific modules can replace :class:`ModuleRecord` with a dedicated model. +""" + +from typing import Literal, Optional + +from pydantic import BaseModel, ConfigDict, JsonValue, RootModel + +OUTPUT_SCHEMA_VERSION: Literal["1.0"] = "1.0" +OutputModel = type[BaseModel] + + +class MVTOutputModel(BaseModel): + """Base for stable output objects with no undocumented fields.""" + + model_config = ConfigDict(extra="forbid", strict=True) + + +class ModuleRecord(BaseModel): + """A JSON object produced by an extraction module. + + Module-specific keys are retained and represented in JSON Schema through + ``additionalProperties``. Dedicated module models can provide stronger field + contracts without changing the serialization machinery. + """ + + model_config = ConfigDict(extra="allow", strict=True) + + +class RecordListOutput(RootModel[list[ModuleRecord]]): + """The standard output shape for built-in extraction modules.""" + + +class MappingOutput(RootModel[dict[str, JsonValue]]): + """Output shape for modules grouped by a dynamic string key.""" + + +class GenericModuleOutput(RootModel[list[JsonValue] | dict[str, JsonValue]]): + """Compatibility output shape for third-party modules without a model.""" + + +class FileHash(MVTOutputModel): + file_path: str + sha256: str + + +class RunInfo(MVTOutputModel): + target_path: Optional[str] + mvt_version: str + date: str + ioc_files: list[str] + hashes: list[FileHash] + output_schema_version: Literal["1.0"] + + +class URLResult(MVTOutputModel): + url: str + expanded_url: Optional[str] + timestamp: Optional[str] + source: str + + +class URLResults(RootModel[list[URLResult]]): + pass + + +AlertLevelName = Literal["INFORMATIONAL", "LOW", "MEDIUM", "HIGH", "CRITICAL"] + + +class AlertResult(MVTOutputModel): + level: AlertLevelName + module: str + message: str + event_time: str + event: dict[str, JsonValue] + matched_indicator: Optional[JsonValue] = None + + +class AlertResults(RootModel[list[AlertResult]]): + pass + + +class TimelineEvent(MVTOutputModel): + """Public contract for the records used to create ``timeline.csv``.""" + + timestamp: Optional[str] + module: str + event: str + data: str + + +class TimelineResults(RootModel[list[TimelineEvent]]): + pass diff --git a/src/mvt/schemas/registry.py b/src/mvt/schemas/registry.py new file mode 100644 index 0000000..a9df335 --- /dev/null +++ b/src/mvt/schemas/registry.py @@ -0,0 +1,133 @@ +# Mobile Verification Toolkit (MVT) +# Copyright (c) 2026 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/ + +"""Discovery and JSON Schema export for MVT output models.""" + +import json +import os +from pathlib import Path +from typing import TYPE_CHECKING, Any, Iterable + +from pydantic import BaseModel + +from .models import ( + OUTPUT_SCHEMA_VERSION, + AlertResults, + RunInfo, + TimelineResults, + URLResults, +) + +if TYPE_CHECKING: + from mvt.common.module import MVTModule + +COMMON_OUTPUT_MODELS: dict[str, type[BaseModel]] = { + "alerts": AlertResults, + "info": RunInfo, + "timeline": TimelineResults, + "urls": URLResults, +} + +JSON_SCHEMA_DIALECT = "https://json-schema.org/draft/2020-12/schema" + + +def _module_classes(platform: str) -> Iterable[type["MVTModule"]]: + if platform == "ios": + from mvt.ios.modules.backup import BACKUP_MODULES as IOS_BACKUP_MODULES + from mvt.ios.modules.fs import FS_MODULES + from mvt.ios.modules.mixed import MIXED_MODULES + + return IOS_BACKUP_MODULES + FS_MODULES + MIXED_MODULES + if platform == "android": + from mvt.android.modules.androidqf import ANDROIDQF_MODULES + from mvt.android.modules.backup import BACKUP_MODULES as ANDROID_BACKUP_MODULES + from mvt.android.modules.bugreport import BUGREPORT_MODULES + from mvt.android.modules.intrusion_logs import INTRUSION_LOGS_MODULES + + return ( + ANDROID_BACKUP_MODULES + + BUGREPORT_MODULES + + ANDROIDQF_MODULES + + INTRUSION_LOGS_MODULES + ) + raise ValueError(f"Unsupported MVT platform: {platform}") + + +def get_output_models(platform: str) -> dict[str, type[BaseModel]]: + """Return common and module output models available for a platform.""" + + models = dict(COMMON_OUTPUT_MODELS) + for module in _module_classes(platform): + models[module.get_slug()] = module.output_model + return dict(sorted(models.items())) + + +def get_output_model(name: str, platform: str | None = None) -> type[BaseModel]: + """Look up an output model by file stem/module slug. + + ``platform`` is optional for common outputs. It is required when an output + slug exists on both platforms and resolves to different models. + """ + + normalized = name.removesuffix(".json") + if normalized.endswith("_detected"): + return AlertResults + if normalized in COMMON_OUTPUT_MODELS: + return COMMON_OUTPUT_MODELS[normalized] + if platform: + try: + return get_output_models(platform)[normalized] + except KeyError as exc: + raise KeyError(f"Unknown {platform} output schema: {name}") from exc + + matches = { + models[normalized] + for candidate in ("ios", "android") + if normalized in (models := get_output_models(candidate)) + } + if len(matches) == 1: + return matches.pop() + if len(matches) > 1: + raise KeyError(f"Output schema {name!r} is ambiguous; specify a platform") + raise KeyError(f"Unknown MVT output schema: {name}") + + +def schema_bundle(platform: str) -> dict[str, Any]: + """Build a versioned bundle of Draft 2020-12 JSON Schemas.""" + + return { + "output_schema_version": OUTPUT_SCHEMA_VERSION, + "platform": platform, + "schemas": { + name: _output_json_schema(name, model, platform) + for name, model in get_output_models(platform).items() + }, + } + + +def _output_json_schema( + name: str, model: type[BaseModel], platform: str +) -> dict[str, Any]: + schema = model.model_json_schema() + return { + "$schema": JSON_SCHEMA_DIALECT, + "$id": f"urn:mvt:output-schema:{OUTPUT_SCHEMA_VERSION}:{platform}:{name}", + **schema, + } + + +def export_json_schemas(destination: str | os.PathLike[str], platform: str) -> list[Path]: + """Write one JSON Schema per output and return the paths created.""" + + destination_path = Path(destination) + destination_path.mkdir(parents=True, exist_ok=True) + written = [] + for name, model in get_output_models(platform).items(): + output_path = destination_path / f"{name}.schema.json" + with output_path.open("w", encoding="utf-8") as handle: + json.dump(_output_json_schema(name, model, platform), handle, indent=2) + handle.write("\n") + written.append(output_path) + return written diff --git a/src/mvt/schemas/serialization.py b/src/mvt/schemas/serialization.py new file mode 100644 index 0000000..d2a91e3 --- /dev/null +++ b/src/mvt/schemas/serialization.py @@ -0,0 +1,35 @@ +# Mobile Verification Toolkit (MVT) +# Copyright (c) 2026 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/ + +"""Validation and serialization helpers shared by all MVT output writers.""" + +import json +from pathlib import Path +from typing import Any + +from pydantic import BaseModel + +from mvt.common.utils import CustomJSONEncoder + + +def json_compatible(value: Any) -> Any: + """Apply MVT's legacy conversions before validating JSON-native values.""" + + return json.loads(json.dumps(value, cls=CustomJSONEncoder)) + + +def validate_output(model: type[BaseModel], value: Any) -> Any: + """Validate an output document and return JSON-serializable Python values.""" + + compatible = json_compatible(value) + return model.model_validate(compatible).model_dump(mode="json") + + +def write_output(path: str | Path, model: type[BaseModel], value: Any) -> None: + """Validate and write an MVT JSON document.""" + + validated = validate_output(model, value) + with open(path, "w", encoding="utf-8") as handle: + json.dump(validated, handle, indent=4) diff --git a/tests/common/test_schemas.py b/tests/common/test_schemas.py new file mode 100644 index 0000000..c99d384 --- /dev/null +++ b/tests/common/test_schemas.py @@ -0,0 +1,131 @@ +# Mobile Verification Toolkit (MVT) +# Copyright (c) 2026 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 pytest +from click.testing import CliRunner +from pydantic import BaseModel, ConfigDict, RootModel, ValidationError + +from mvt.android.cli import cli as android_cli +from mvt.common.module import MVTModule +from mvt.ios.cli import cli as ios_cli +from mvt.schemas import ( + OUTPUT_SCHEMA_VERSION, + GenericModuleOutput, + MappingOutput, + RecordListOutput, + export_json_schemas, + get_output_model, + get_output_models, + schema_bundle, + validate_output, +) + + +class CountRecord(BaseModel): + model_config = ConfigDict(strict=True) + count: int + + +class CountOutput(RootModel[list[CountRecord]]): + pass + + +class CountModule(MVTModule): + output_model = CountOutput + + +def test_common_output_model_validates_without_changing_root_shape(): + assert validate_output(RecordListOutput, [{"name": "example", "size": 2}]) == [ + {"name": "example", "size": 2} + ] + assert validate_output(MappingOutput, {"global": {"enabled": "1"}}) == { + "global": {"enabled": "1"} + } + + +def test_module_model_is_used_when_loading_existing_output(tmp_path): + path = tmp_path / "count.json" + path.write_text('[{"count": 4}]', encoding="utf-8") + + module = CountModule.from_json(str(path), logging.getLogger(__name__)) + + assert module.results == [{"count": 4}] + + +def test_invalid_module_output_is_not_written(tmp_path, caplog): + module = CountModule(results_path=str(tmp_path), results=[{"count": "four"}]) + + with caplog.at_level(logging.ERROR): + module.save_to_json() + + assert not (tmp_path / "count_module.json").exists() + assert "does not match schema CountOutput" in caplog.text + + +def test_invalid_existing_output_raises_validation_error(tmp_path): + path = tmp_path / "count.json" + path.write_text('[{"count": "four"}]', encoding="utf-8") + + with pytest.raises(ValidationError): + CountModule.from_json(str(path), logging.getLogger(__name__)) + + +@pytest.mark.parametrize("platform", ["ios", "android"]) +def test_every_builtin_module_has_a_non_generic_output_model(platform): + module_models = { + name: model + for name, model in get_output_models(platform).items() + if name not in {"alerts", "info", "timeline", "urls"} + } + + assert module_models + assert GenericModuleOutput not in module_models.values() + + +def test_known_mapping_and_specific_models_are_registered(): + assert get_output_model("backup_info.json", "ios") is MappingOutput + assert get_output_model("dumpsys_receivers", "android") is MappingOutput + assert get_output_model("tombstones", "android").__name__ == "TombstoneCrashOutput" + assert get_output_model("sms_detected.json", "ios").__name__ == "AlertResults" + + +def test_schema_bundle_has_stable_versioned_shape(): + bundle = schema_bundle("android") + + assert bundle["output_schema_version"] == OUTPUT_SCHEMA_VERSION + assert bundle["platform"] == "android" + assert bundle["schemas"]["urls"]["$schema"].endswith("2020-12/schema") + assert bundle["schemas"]["urls"]["type"] == "array" + assert bundle["schemas"]["info"]["additionalProperties"] is False + assert bundle["schemas"]["dumpsys_receivers"]["type"] == "object" + + +def test_export_json_schemas_writes_one_file_per_registered_output(tmp_path): + paths = export_json_schemas(tmp_path, "android") + + assert len(paths) == len(get_output_models("android")) + assert json.loads((tmp_path / "info.schema.json").read_text())["title"] == ( + "RunInfo" + ) + + +@pytest.mark.parametrize( + ("cli", "platform"), [(ios_cli, "ios"), (android_cli, "android")] +) +def test_schema_cli_prints_machine_readable_bundle(cli, platform): + result = CliRunner().invoke(cli, ["schemas"]) + + assert result.exit_code == 0 + assert json.loads(result.output)["platform"] == platform + + +def test_schema_cli_exports_schema_files(tmp_path): + result = CliRunner().invoke(android_cli, ["schemas", "--output", str(tmp_path)]) + + assert result.exit_code == 0 + assert (tmp_path / "urls.schema.json").exists()