mirror of
https://github.com/mvt-project/mvt.git
synced 2026-10-29 17:26:33 +01:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
895307e242 | ||
|
|
e5a9a1f919 | ||
|
|
c5d7f5a1d0 |
No files matched your search
@@ -62,6 +62,27 @@ AndroidQF will prompt the user to download, decrypt and collect device intrusion
|
||||
|
||||
For cases where intrusion logs were collected outside of an AndroidQF acquisition, the standalone `mvt-android check-intrusion-logs` command can analyse them directly. See [Check Android Intrusion Logs](intrusion_logs.md) for details, and the [feature announcment from Amnesty International's Security Lab](https://securitylab.amnesty.org/latest/2026/05/android-intrusion-logging-as-a-new-source-of-data-for-consensual-forensic-analysis/) for background on the data source.
|
||||
|
||||
## Browser history
|
||||
|
||||
AndroidQF can optionally collect browser `History` databases when a device
|
||||
already has working root access. When its acquisition contains the
|
||||
`browser_history/manifest.json` file, `mvt-android check-androidqf` parses every
|
||||
listed database, including any collected SQLite WAL and SHM sidecars. Browser
|
||||
visits are added to the module results, timeline, URL output, and indicator
|
||||
checks.
|
||||
|
||||
MVT can also inspect the supported database locations in a full Android
|
||||
filesystem dump, or inspect one explicitly supplied Chromium `History` file:
|
||||
|
||||
```bash
|
||||
mvt-android check-fs --output /path/to/results /path/to/filesystem-dump
|
||||
mvt-android check-fs --output /path/to/results /path/to/History
|
||||
```
|
||||
|
||||
The built-in locations cover Chrome, Brave, Microsoft Edge, and Samsung
|
||||
Internet. Other Chromium-compatible databases can be supplied explicitly as a
|
||||
file without assigning an unverified browser identity.
|
||||
|
||||
## Android Debug Bridge analysis removed
|
||||
|
||||
The ability to analyze Android devices directly over ADB has been removed from MVT. Direct extraction of data from ADB was error-prone and frequently resulted in inconsistent data collection between ADB and AndroidQF acquisitions. Use AndroidQF for device acquisition and `mvt-android check-androidqf` for analysis.
|
||||
|
||||
@@ -1,16 +0,0 @@
|
||||
# Coruna and DarkSword artifacts
|
||||
|
||||
The Manifest and Filesystem modules flag two scoped Coruna path patterns without requiring an IOC file:
|
||||
|
||||
- `com.apple.<16 hexadecimal characters>.plist` in the mobile user's `Library/Preferences` directory (the `HomeDomain` equivalent in a backup).
|
||||
- `Documents/.devcache` at the root of an application data container (`AppDomain-<bundle ID>` in a backup).
|
||||
|
||||
Path-only matches produce medium-severity alerts. When file contents are available, the modules read at most 1 MiB and check the structures published in [iVerify's September 2026 report](https://www.iverify.com/blog/proliferation-of-coruna-and-darksword). A photo-tracking plist must contain a DCIM path and a positive finite processing timestamp. A device cache must identify `exp` as `plasma` and contain nonempty `ecid`, `gci`, and `unique-id` strings. Matching structures produce high-severity alerts. Matching device identifiers in distinct application containers produce an additional correlation alert. Results retain a fingerprint for correlation instead of raw device identifiers or the tracked photo path.
|
||||
|
||||
File content checks accept binary/XML plists and JSON. Missing, malformed or oversized content does not erase a path finding. A path finding alone is not confirmation of compromise. The checks do not inspect arbitrary plists based solely on field names, or flag `.devcache` outside the documented application-container location.
|
||||
|
||||
Backup domains with known fixed roots are mapped to device paths for IOC matching. Application container UUIDs are not invented from bundle IDs. iOS path comparisons handle `/var` versus `/private/var`, `/tmp` versus `/private/tmp`, and host path separators. File path indicators match the exact path or descendants at a directory boundary, not similarly prefixed sibling names.
|
||||
|
||||
For payload hashes, pass `--hashes` to `check-fs` or `check-backup` (or enable `HASH_FILES` in the configuration). The Filesystem and Manifest modules then store SHA-256 values and compare them with loaded hash IOCs; the saved module results can also be rechecked later. This reads available file contents and increases processing time. Acquisition integrity hashes in `info.json` remain available as before. Reads for artifact inspection and these module hashes are confined to regular files inside the acquisition; symlinks are not followed.
|
||||
|
||||
Decrypt encrypted backups before analysis. Sysdiagnose and filesystem metadata may expose paths without the corresponding file bytes, and absence from a partial acquisition does not establish absence from the device. Changes to payload bytes defeat exact hash matching; source-level sample names and legitimate injected process names are not standalone malware indicators.
|
||||
@@ -0,0 +1,162 @@
|
||||
# Mobile Verification Toolkit (MVT)
|
||||
# Copyright (c) 2021-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 sqlite3
|
||||
import shutil
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from mvt.common.module_types import ModuleAtomicResult, ModuleSerializedResult
|
||||
from mvt.common.utils import convert_chrometime_to_datetime, convert_datetime_to_iso
|
||||
|
||||
from .artifact import AndroidArtifact
|
||||
|
||||
|
||||
class BrowserHistoryArtifact(AndroidArtifact):
|
||||
"""Shared Chromium History database parsing and result handling."""
|
||||
|
||||
def _parse_browser_history(
|
||||
self,
|
||||
connection: sqlite3.Connection,
|
||||
*,
|
||||
browser: str,
|
||||
package: str,
|
||||
profile: str,
|
||||
source_path: str,
|
||||
) -> None:
|
||||
cursor = connection.cursor()
|
||||
try:
|
||||
cursor.execute(
|
||||
"""
|
||||
SELECT
|
||||
urls.id,
|
||||
urls.url,
|
||||
urls.title,
|
||||
urls.visit_count,
|
||||
urls.typed_count,
|
||||
visits.id,
|
||||
visits.visit_time,
|
||||
visits.from_visit,
|
||||
visits.transition
|
||||
FROM urls
|
||||
JOIN visits ON visits.url = urls.id
|
||||
ORDER BY visits.visit_time;
|
||||
"""
|
||||
)
|
||||
for row in cursor:
|
||||
timestamp = int(row[6])
|
||||
self.results.append(
|
||||
{
|
||||
"id": row[0],
|
||||
"url": row[1],
|
||||
"title": row[2],
|
||||
"visit_count": row[3],
|
||||
"typed_count": row[4],
|
||||
"visit_id": row[5],
|
||||
"timestamp": timestamp,
|
||||
"isodate": convert_datetime_to_iso(
|
||||
convert_chrometime_to_datetime(timestamp)
|
||||
),
|
||||
"redirect_source": row[7],
|
||||
"transition": row[8],
|
||||
"browser": browser,
|
||||
"package": package,
|
||||
"profile": profile,
|
||||
"source_path": source_path,
|
||||
}
|
||||
)
|
||||
finally:
|
||||
cursor.close()
|
||||
|
||||
def serialize(self, record: ModuleAtomicResult) -> ModuleSerializedResult:
|
||||
return {
|
||||
"timestamp": record["isodate"],
|
||||
"module": self.__class__.__name__,
|
||||
"event": "browser_history",
|
||||
"data": (
|
||||
f"{record['browser']} visit to {record['url']} "
|
||||
f"(visit ID: {record['visit_id']}, profile: {record['profile']})"
|
||||
),
|
||||
}
|
||||
|
||||
def check_indicators(self) -> None:
|
||||
if not self.indicators:
|
||||
return
|
||||
|
||||
for result, match in zip(
|
||||
self.results,
|
||||
self.indicators.check_url_batches(
|
||||
[[result["url"]] for result in self.results]
|
||||
),
|
||||
):
|
||||
if match:
|
||||
self.alertstore.critical(
|
||||
match.message, "", result, matched_indicator=match.ioc
|
||||
)
|
||||
|
||||
def collect_url_results(self) -> None:
|
||||
for result in self.results:
|
||||
self.add_url_result(result["url"], result.get("isodate"), "browser_history")
|
||||
|
||||
|
||||
class TemporarySQLiteConnection(sqlite3.Connection):
|
||||
temporary_directory: tempfile.TemporaryDirectory | None = None
|
||||
|
||||
def close(self) -> None:
|
||||
try:
|
||||
super().close()
|
||||
finally:
|
||||
if self.temporary_directory:
|
||||
self.temporary_directory.cleanup()
|
||||
self.temporary_directory = None
|
||||
|
||||
|
||||
def open_browser_history_database(database_path: Path) -> sqlite3.Connection:
|
||||
"""Open a staged History database without modifying forensic evidence."""
|
||||
database_uri = database_path.resolve().as_uri()
|
||||
if not Path(f"{database_path}-wal").is_file():
|
||||
return sqlite3.connect(f"{database_uri}?mode=ro&immutable=1", uri=True)
|
||||
|
||||
temporary_directory = tempfile.TemporaryDirectory(prefix="mvt_sqlite_")
|
||||
temporary_path = Path(temporary_directory.name) / database_path.name
|
||||
shutil.copy2(database_path, temporary_path)
|
||||
for suffix in ("-wal", "-shm"):
|
||||
sidecar = Path(f"{database_path}{suffix}")
|
||||
if sidecar.is_file():
|
||||
shutil.copy2(sidecar, Path(f"{temporary_path}{suffix}"))
|
||||
|
||||
try:
|
||||
connection = sqlite3.connect(
|
||||
f"{temporary_path.resolve().as_uri()}?mode=ro",
|
||||
uri=True,
|
||||
factory=TemporarySQLiteConnection,
|
||||
)
|
||||
except Exception:
|
||||
temporary_directory.cleanup()
|
||||
raise
|
||||
connection.temporary_directory = temporary_directory
|
||||
return connection
|
||||
|
||||
|
||||
def validate_manifest_database(database: Any) -> dict[str, Any]:
|
||||
if not isinstance(database, dict):
|
||||
raise ValueError("database entry is not an object")
|
||||
|
||||
required = ("browser", "package", "profile", "device_path", "archive_path")
|
||||
for field in required:
|
||||
if not isinstance(database.get(field), str) or not database[field]:
|
||||
raise ValueError(f"database entry has invalid {field}")
|
||||
|
||||
archive_path = database["archive_path"]
|
||||
path = Path(archive_path)
|
||||
if (
|
||||
"\\" in archive_path
|
||||
or path.is_absolute()
|
||||
or ".." in path.parts
|
||||
or path.parts[:1] != ("browser_history",)
|
||||
):
|
||||
raise ValueError(f"unsafe browser history archive path: {archive_path}")
|
||||
return database
|
||||
@@ -21,6 +21,7 @@ from mvt.common.help import (
|
||||
HELP_MSG_CHECK_ANDROID_BACKUP,
|
||||
HELP_MSG_CHECK_ANDROIDQF,
|
||||
HELP_MSG_CHECK_BUGREPORT,
|
||||
HELP_MSG_CHECK_FS,
|
||||
HELP_MSG_CHECK_IOCS,
|
||||
HELP_MSG_CHECK_INTRUSION_LOGS,
|
||||
HELP_MSG_DELAY_CHECKS,
|
||||
@@ -361,6 +362,59 @@ def check_androidqf(
|
||||
cmd.show_support_message()
|
||||
|
||||
|
||||
# ==============================================================================
|
||||
# Command: check-fs
|
||||
# ==============================================================================
|
||||
@cli.command("check-fs", context_settings=CONTEXT_SETTINGS, help=HELP_MSG_CHECK_FS)
|
||||
@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("DUMP_PATH", type=click.Path(exists=True))
|
||||
@click.pass_context
|
||||
def check_fs(
|
||||
ctx, iocs, output, list_modules, module, load_module, hashes, verbose, dump_path
|
||||
):
|
||||
set_verbose_logging(verbose)
|
||||
custom_modules = _load_custom_modules(load_module)
|
||||
from .cmd_check_fs import CmdAndroidCheckFS
|
||||
|
||||
cmd = CmdAndroidCheckFS(
|
||||
target_path=dump_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 list_modules:
|
||||
cmd.list_modules()
|
||||
return
|
||||
|
||||
log.info("Checking Android filesystem located at: %s", dump_path)
|
||||
cmd.run()
|
||||
cmd.show_alerts_brief()
|
||||
cmd.show_support_message()
|
||||
|
||||
|
||||
# ==============================================================================
|
||||
# Command: check-intrusion-logs
|
||||
# ==============================================================================
|
||||
|
||||
@@ -0,0 +1,47 @@
|
||||
# Mobile Verification Toolkit (MVT)
|
||||
# Copyright (c) 2021-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 logging
|
||||
from typing import Optional
|
||||
|
||||
from mvt.common.command import Command
|
||||
from mvt.common.indicators import Indicators
|
||||
from mvt.common.module import MVTModule
|
||||
|
||||
from .modules.fs import FS_MODULES
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class CmdAndroidCheckFS(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,
|
||||
module_options: Optional[dict] = None,
|
||||
hashes: 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,
|
||||
module_options=module_options,
|
||||
hashes=hashes,
|
||||
log=log,
|
||||
disable_version_check=disable_version_check,
|
||||
disable_indicator_check=disable_indicator_check,
|
||||
custom_modules=custom_modules,
|
||||
)
|
||||
self.platform = "android"
|
||||
self.name = "check-fs"
|
||||
self.modules = FS_MODULES
|
||||
@@ -11,6 +11,7 @@ from .aqf_log_timestamps import AQFLogTimestamps
|
||||
from .aqf_packages import AQFPackages
|
||||
from .aqf_processes import AQFProcesses
|
||||
from .aqf_settings import AQFSettings
|
||||
from .browser_history import BrowserHistory
|
||||
from .mounts import Mounts
|
||||
from .root_binaries import RootBinaries
|
||||
|
||||
@@ -23,4 +24,5 @@ ANDROIDQF_MODULES: list[type[MVTModule]] = [
|
||||
AQFLogTimestamps,
|
||||
RootBinaries,
|
||||
Mounts,
|
||||
BrowserHistory,
|
||||
]
|
||||
@@ -0,0 +1,115 @@
|
||||
# Mobile Verification Toolkit (MVT)
|
||||
# Copyright (c) 2021-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 sqlite3
|
||||
import tempfile
|
||||
from pathlib import Path, PurePosixPath
|
||||
|
||||
from mvt.android.artifacts.browser_history import (
|
||||
BrowserHistoryArtifact,
|
||||
open_browser_history_database,
|
||||
validate_manifest_database,
|
||||
)
|
||||
|
||||
from .base import AndroidQFModule
|
||||
|
||||
|
||||
class BrowserHistory(BrowserHistoryArtifact, AndroidQFModule):
|
||||
"""Extract browser visits collected by AndroidQF."""
|
||||
|
||||
supported_commands = (("android", "check-androidqf"),)
|
||||
|
||||
def _find_manifest(self) -> str | None:
|
||||
manifests = [
|
||||
file_path
|
||||
for file_path in self.files
|
||||
if file_path.replace("\\", "/").endswith("browser_history/manifest.json")
|
||||
]
|
||||
if not manifests:
|
||||
return None
|
||||
if len(manifests) > 1:
|
||||
self.log.warning(
|
||||
"Found multiple browser history manifests; using %s", manifests[0]
|
||||
)
|
||||
return manifests[0]
|
||||
|
||||
def _stage_database(
|
||||
self, archive_path: str, prefix: str, temporary_path: Path
|
||||
) -> Path:
|
||||
available_files = {
|
||||
str(PurePosixPath(file_path.replace("\\", "/"))): file_path
|
||||
for file_path in self.files
|
||||
}
|
||||
normalized_path = str(PurePosixPath(prefix, archive_path))
|
||||
source_path = available_files.get(normalized_path)
|
||||
if not source_path:
|
||||
raise FileNotFoundError(archive_path)
|
||||
|
||||
staged_path = temporary_path / "History"
|
||||
staged_path.write_bytes(self._get_file_content(source_path))
|
||||
for suffix in ("-wal", "-shm"):
|
||||
sidecar = available_files.get(normalized_path + suffix)
|
||||
if sidecar:
|
||||
Path(f"{staged_path}{suffix}").write_bytes(
|
||||
self._get_file_content(sidecar)
|
||||
)
|
||||
return staged_path
|
||||
|
||||
def run(self) -> None:
|
||||
manifest_path = self._find_manifest()
|
||||
if not manifest_path:
|
||||
self.log.info("No AndroidQF browser history manifest found")
|
||||
return
|
||||
|
||||
try:
|
||||
manifest = json.loads(self._get_file_content(manifest_path))
|
||||
except (json.JSONDecodeError, OSError, TypeError, UnicodeDecodeError) as exc:
|
||||
self.log.error("Unable to read browser history manifest: %s", exc)
|
||||
return
|
||||
|
||||
if not isinstance(manifest, dict) or manifest.get("schema_version") != 1:
|
||||
self.log.error("Unsupported AndroidQF browser history manifest")
|
||||
return
|
||||
databases = manifest.get("databases", [])
|
||||
if not isinstance(databases, list):
|
||||
self.log.error("Invalid AndroidQF browser history database list")
|
||||
return
|
||||
|
||||
normalized_manifest = manifest_path.replace("\\", "/")
|
||||
marker = "browser_history/manifest.json"
|
||||
prefix = normalized_manifest[: -len(marker)].rstrip("/")
|
||||
|
||||
for raw_database in databases:
|
||||
try:
|
||||
database = validate_manifest_database(raw_database)
|
||||
with tempfile.TemporaryDirectory(prefix="mvt_browser_history_") as temp:
|
||||
staged_path = self._stage_database(
|
||||
database["archive_path"], prefix, Path(temp)
|
||||
)
|
||||
connection = open_browser_history_database(staged_path)
|
||||
try:
|
||||
self._parse_browser_history(
|
||||
connection,
|
||||
browser=database["browser"],
|
||||
package=database["package"],
|
||||
profile=database["profile"],
|
||||
source_path=database["device_path"],
|
||||
)
|
||||
finally:
|
||||
connection.close()
|
||||
except (
|
||||
FileNotFoundError,
|
||||
OSError,
|
||||
OverflowError,
|
||||
sqlite3.Error,
|
||||
TypeError,
|
||||
ValueError,
|
||||
) as exc:
|
||||
self.log.error("Unable to parse browser history database: %s", exc)
|
||||
|
||||
self.log.info(
|
||||
"Extracted a total of %d browser history items", len(self.results)
|
||||
)
|
||||
@@ -0,0 +1,10 @@
|
||||
# Mobile Verification Toolkit (MVT)
|
||||
# Copyright (c) 2021-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/
|
||||
|
||||
from mvt.common.module import MVTModule
|
||||
|
||||
from .browser_history import BrowserHistory
|
||||
|
||||
FS_MODULES: list[type[MVTModule]] = [BrowserHistory]
|
||||
@@ -0,0 +1,115 @@
|
||||
# Mobile Verification Toolkit (MVT)
|
||||
# Copyright (c) 2021-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 logging
|
||||
import sqlite3
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from mvt.android.artifacts.browser_history import (
|
||||
BrowserHistoryArtifact,
|
||||
open_browser_history_database,
|
||||
)
|
||||
from mvt.common.module import MVTModule
|
||||
from mvt.common.module_types import ModuleResults
|
||||
|
||||
|
||||
# These locations are deliberately limited to paths backed by public parser
|
||||
# fixtures or the historical MVT implementation.
|
||||
BROWSER_HISTORY_PATHS = {
|
||||
"data/data/com.android.chrome/app_chrome/Default/History": (
|
||||
"Chrome",
|
||||
"com.android.chrome",
|
||||
"Default",
|
||||
),
|
||||
"data/data/com.brave.browser/app_chrome/Default/History": (
|
||||
"Brave",
|
||||
"com.brave.browser",
|
||||
"Default",
|
||||
),
|
||||
"data/data/com.microsoft.emmx/app_chrome/Default/History": (
|
||||
"Microsoft Edge",
|
||||
"com.microsoft.emmx",
|
||||
"Default",
|
||||
),
|
||||
"data/data/com.sec.android.app.sbrowser/app_sbrowser/Default/History": (
|
||||
"Samsung Internet",
|
||||
"com.sec.android.app.sbrowser",
|
||||
"Default",
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
class BrowserHistory(BrowserHistoryArtifact, MVTModule):
|
||||
"""Extract supported Chromium History databases from a filesystem dump."""
|
||||
|
||||
supported_commands = (("android", "check-fs"),)
|
||||
|
||||
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,
|
||||
)
|
||||
|
||||
def _database_paths(self) -> list[tuple[Path, str, str, str]]:
|
||||
if not self.target_path:
|
||||
return []
|
||||
target = Path(self.target_path)
|
||||
if target.is_file():
|
||||
normalized_target = target.resolve().as_posix()
|
||||
for relative_path, identity in BROWSER_HISTORY_PATHS.items():
|
||||
if normalized_target.endswith(f"/{relative_path}"):
|
||||
return [(target, *identity)]
|
||||
return [(target, "Chromium", "unknown", "unknown")]
|
||||
|
||||
databases = []
|
||||
for relative_path, identity in BROWSER_HISTORY_PATHS.items():
|
||||
database_path = target / relative_path
|
||||
if database_path.is_file():
|
||||
databases.append((database_path, *identity))
|
||||
return databases
|
||||
|
||||
def run(self) -> None:
|
||||
for database_path, browser, package, profile in self._database_paths():
|
||||
try:
|
||||
connection = open_browser_history_database(database_path)
|
||||
try:
|
||||
self._parse_browser_history(
|
||||
connection,
|
||||
browser=browser,
|
||||
package=package,
|
||||
profile=profile,
|
||||
source_path=str(database_path),
|
||||
)
|
||||
finally:
|
||||
connection.close()
|
||||
except (
|
||||
OSError,
|
||||
OverflowError,
|
||||
sqlite3.Error,
|
||||
TypeError,
|
||||
ValueError,
|
||||
) as exc:
|
||||
self.log.error(
|
||||
"Unable to parse browser history database %s: %s",
|
||||
database_path,
|
||||
exc,
|
||||
)
|
||||
|
||||
self.log.info(
|
||||
"Extracted a total of %d browser history items", len(self.results)
|
||||
)
|
||||
@@ -646,19 +646,14 @@ class Indicators:
|
||||
if not file_path:
|
||||
return None
|
||||
|
||||
# Device paths use forward slashes even when analysed on Windows.
|
||||
file_path = file_path.replace("\\", "/")
|
||||
ioc_match = self.check_file_name(file_path.rsplit("/", 1)[-1])
|
||||
ioc_match = self.check_file_name(os.path.basename(file_path))
|
||||
if ioc_match:
|
||||
return ioc_match
|
||||
|
||||
for ioc in self.get_iocs("file_paths"):
|
||||
# Strip any trailing slash from indicator paths to match
|
||||
# directories.
|
||||
indicator_path = ioc.value.replace("\\", "/").rstrip("/")
|
||||
if file_path == indicator_path or file_path.startswith(
|
||||
indicator_path + "/"
|
||||
):
|
||||
if file_path.startswith(ioc.value.rstrip("/")):
|
||||
return IndicatorMatch(
|
||||
ioc=ioc,
|
||||
message=f'Found a known suspicious file path "{file_path}" matching indicators form "{ioc.name}"',
|
||||
|
||||
@@ -8,7 +8,6 @@ import os
|
||||
from typing import Optional
|
||||
|
||||
from mvt.common.command import Command
|
||||
from mvt.common.config import settings
|
||||
from mvt.common.indicators import Indicators
|
||||
from mvt.common.module import MVTModule
|
||||
|
||||
@@ -57,8 +56,6 @@ class CmdIOSCheckBackup(Command):
|
||||
)
|
||||
|
||||
self.platform = "ios"
|
||||
if hashes or settings.HASH_FILES:
|
||||
self.module_options["check_file_hashes"] = True
|
||||
self.name = "check-backup"
|
||||
self.modules = BACKUP_MODULES + MIXED_MODULES
|
||||
|
||||
|
||||
@@ -7,7 +7,6 @@ import logging
|
||||
from typing import Optional
|
||||
|
||||
from mvt.common.command import Command
|
||||
from mvt.common.config import settings
|
||||
from mvt.common.indicators import Indicators
|
||||
from mvt.common.module import MVTModule
|
||||
|
||||
@@ -49,8 +48,6 @@ class CmdIOSCheckFS(Command):
|
||||
)
|
||||
|
||||
self.platform = "ios"
|
||||
if hashes or settings.HASH_FILES:
|
||||
self.module_options["check_file_hashes"] = True
|
||||
self.name = "check-fs"
|
||||
self.modules = FS_MODULES + MIXED_MODULES
|
||||
|
||||
|
||||
@@ -1,168 +0,0 @@
|
||||
"""Coruna artifact checks shared by backups, filesystems and metadata listings.
|
||||
|
||||
Structures documented by iVerify on 2026-09-15:
|
||||
https://www.iverify.com/blog/proliferation-of-coruna-and-darksword
|
||||
"""
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import math
|
||||
import plistlib
|
||||
import re
|
||||
import stat
|
||||
from collections import defaultdict
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
from xml.parsers.expat import ExpatError
|
||||
|
||||
from mvt.common.alerts import AlertStore
|
||||
|
||||
from .paths import backup_device_path, normalize_ios_path
|
||||
|
||||
_PHOTO_STATE = re.compile(
|
||||
r"/private/var/mobile/Library/Preferences/com\.apple\.[0-9a-fA-F]{16}\.plist"
|
||||
)
|
||||
_DEVICE_CACHE = re.compile(
|
||||
r"/private/var/mobile/Containers/Data/Application/"
|
||||
r"(?P<container>[0-9a-fA-F]{8}(?:-[0-9a-fA-F]{4}){3}-[0-9a-fA-F]{12})"
|
||||
r"/Documents/\.devcache"
|
||||
)
|
||||
MAX_ARTIFACT_SIZE = 1024 * 1024
|
||||
|
||||
|
||||
def coruna_path_artifact(path: str, domain: Optional[str] = None) -> Optional[dict]:
|
||||
"""Return a scoped candidate, never a verdict based on a basename alone."""
|
||||
if domain is not None:
|
||||
relative = path.replace("\\", "/")
|
||||
if relative == "Documents/.devcache" and domain.startswith("AppDomain-"):
|
||||
return {"kind": "device-cache", "container": domain}
|
||||
path = backup_device_path(domain, relative) or ""
|
||||
if not path:
|
||||
return None
|
||||
path = normalize_ios_path(path)
|
||||
if _PHOTO_STATE.fullmatch(path):
|
||||
return {"kind": "photo-state"}
|
||||
match = _DEVICE_CACHE.fullmatch(path)
|
||||
if match:
|
||||
return {"kind": "device-cache", "container": match["container"].lower()}
|
||||
return None
|
||||
|
||||
|
||||
def inspect_coruna_artifact(file_path: str, artifact: dict) -> dict:
|
||||
"""Enrich a candidate without exporting photo paths or device identifiers."""
|
||||
result = dict(artifact)
|
||||
try:
|
||||
with open(file_path, "rb") as handle:
|
||||
raw = handle.read(MAX_ARTIFACT_SIZE + 1)
|
||||
if len(raw) > MAX_ARTIFACT_SIZE:
|
||||
return {**result, "content_status": "too-large"}
|
||||
try:
|
||||
data = plistlib.loads(raw)
|
||||
except (ValueError, plistlib.InvalidFileException, ExpatError):
|
||||
data = json.loads(raw)
|
||||
except (OSError, ValueError, UnicodeError, OverflowError):
|
||||
return {**result, "content_status": "unavailable"}
|
||||
|
||||
matched = False
|
||||
if isinstance(data, dict):
|
||||
if artifact["kind"] == "photo-state":
|
||||
photo = data.get("LastProcessedFilePath")
|
||||
timestamp = data.get("LastProcessedTimestamp")
|
||||
matched = (
|
||||
isinstance(photo, str)
|
||||
and normalize_ios_path(photo).startswith(
|
||||
"/private/var/mobile/Media/DCIM/"
|
||||
)
|
||||
and isinstance(timestamp, (int, float))
|
||||
and not isinstance(timestamp, bool)
|
||||
and (isinstance(timestamp, int) or math.isfinite(timestamp))
|
||||
and timestamp > 0
|
||||
)
|
||||
elif artifact["kind"] == "device-cache":
|
||||
keys = ("ecid", "gci", "unique-id")
|
||||
matched = data.get("exp") == "plasma" and all(
|
||||
isinstance(data.get(key), str) and data[key].strip() for key in keys
|
||||
)
|
||||
if matched:
|
||||
identifiers = {key: data[key] for key in keys}
|
||||
result["identifier_sha256"] = hashlib.sha256(
|
||||
json.dumps(identifiers, sort_keys=True).encode()
|
||||
).hexdigest()
|
||||
result["content_status"] = "matched" if matched else "unmatched"
|
||||
return result
|
||||
|
||||
|
||||
def enrich_coruna_record(record: dict, file_path: Optional[str]) -> None:
|
||||
artifact = coruna_path_artifact(
|
||||
record.get("relative_path", record.get("path", "")), record.get("domain")
|
||||
)
|
||||
if artifact:
|
||||
record["coruna_artifact"] = (
|
||||
inspect_coruna_artifact(file_path, artifact) if file_path else artifact
|
||||
)
|
||||
|
||||
|
||||
def alert_coruna_record(record: dict, alertstore: AlertStore) -> None:
|
||||
if record.get("is_directory") or record.get("flags") == 2:
|
||||
return
|
||||
mode = record.get("mode")
|
||||
if isinstance(mode, int) and not stat.S_ISREG(mode):
|
||||
return
|
||||
artifact = record.get("coruna_artifact")
|
||||
if not artifact:
|
||||
artifact = coruna_path_artifact(
|
||||
record.get("relative_path", record.get("path", "")), record.get("domain")
|
||||
)
|
||||
if not artifact:
|
||||
return
|
||||
path = record.get("relative_path", record.get("path", ""))
|
||||
timestamp = record.get("modified", record.get("isodate", ""))
|
||||
if artifact.get("content_status") == "matched":
|
||||
alertstore.high(
|
||||
f"Coruna {artifact['kind']} artifact with matching content structure: {path}",
|
||||
timestamp,
|
||||
record,
|
||||
)
|
||||
else:
|
||||
alertstore.medium(
|
||||
f"Possible Coruna {artifact['kind']} artifact at {path}; "
|
||||
"path match without corroborating content",
|
||||
timestamp,
|
||||
record,
|
||||
)
|
||||
|
||||
|
||||
def correlate_coruna_records(records: list, alertstore: AlertStore) -> None:
|
||||
groups: dict[str, dict[str, dict]] = defaultdict(dict)
|
||||
for record in records:
|
||||
artifact = record.get("coruna_artifact", {})
|
||||
if (
|
||||
artifact.get("kind") != "device-cache"
|
||||
or artifact.get("content_status") != "matched"
|
||||
or not artifact.get("container")
|
||||
):
|
||||
continue
|
||||
fingerprint = artifact.get("identifier_sha256")
|
||||
if fingerprint:
|
||||
groups[fingerprint][artifact["container"]] = record
|
||||
for fingerprint, containers in groups.items():
|
||||
if len(containers) < 2:
|
||||
continue
|
||||
alertstore.high(
|
||||
f"Matching Coruna device identifiers in {len(containers)} application containers",
|
||||
"",
|
||||
{"identifier_sha256": fingerprint, "containers": sorted(containers)},
|
||||
)
|
||||
|
||||
|
||||
def contained_regular_file(root: str, file_path: str) -> bool:
|
||||
"""Do not read symlinks or paths outside the supplied acquisition."""
|
||||
path = Path(file_path)
|
||||
try:
|
||||
return (
|
||||
not path.is_symlink()
|
||||
and path.is_file()
|
||||
and path.resolve().is_relative_to(Path(root).resolve())
|
||||
)
|
||||
except (OSError, RuntimeError, ValueError):
|
||||
return False
|
||||
@@ -18,19 +18,7 @@ from mvt.common.module_types import (
|
||||
ModuleSerializedResult,
|
||||
)
|
||||
from mvt.common.url import URL
|
||||
from mvt.common.utils import (
|
||||
convert_datetime_to_iso,
|
||||
convert_unix_to_iso,
|
||||
get_sha256_from_file_path,
|
||||
)
|
||||
from mvt.ios.coruna import (
|
||||
alert_coruna_record,
|
||||
contained_regular_file,
|
||||
correlate_coruna_records,
|
||||
coruna_path_artifact,
|
||||
enrich_coruna_record,
|
||||
)
|
||||
from mvt.ios.paths import backup_device_path, check_ios_path
|
||||
from mvt.common.utils import convert_datetime_to_iso, convert_unix_to_iso
|
||||
|
||||
from ..base import IOSExtraction
|
||||
|
||||
@@ -106,34 +94,14 @@ class Manifest(IOSExtraction):
|
||||
return records
|
||||
|
||||
def check_indicators(self) -> None:
|
||||
correlate_coruna_records(self.results, self.alertstore)
|
||||
for result in self.results:
|
||||
if not result.get("relative_path"):
|
||||
continue
|
||||
|
||||
alert_coruna_record(result, self.alertstore)
|
||||
if not self.indicators:
|
||||
continue
|
||||
|
||||
hash_match = self.indicators.check_file_hash(result.get("sha256", ""))
|
||||
if hash_match:
|
||||
self.alertstore.critical(
|
||||
hash_match.message,
|
||||
result.get("modified", ""),
|
||||
result,
|
||||
matched_indicator=hash_match.ioc,
|
||||
)
|
||||
|
||||
device_path = backup_device_path(
|
||||
result.get("domain", ""), result["relative_path"]
|
||||
)
|
||||
ioc_match = (
|
||||
check_ios_path(self.indicators, device_path) if device_path else None
|
||||
)
|
||||
if not ioc_match:
|
||||
ioc_match = self.indicators.check_file_path(
|
||||
"/" + result["relative_path"]
|
||||
)
|
||||
ioc_match = self.indicators.check_file_path("/" + result["relative_path"])
|
||||
if ioc_match:
|
||||
self.alertstore.high(
|
||||
ioc_match.message, "", result, matched_indicator=ioc_match.ioc
|
||||
@@ -245,21 +213,6 @@ class Manifest(IOSExtraction):
|
||||
file_data["relativePath"],
|
||||
)
|
||||
|
||||
inspect_content = coruna_path_artifact(
|
||||
cleaned_metadata["relative_path"], cleaned_metadata["domain"]
|
||||
)
|
||||
hash_content = self.module_options.get("check_file_hashes", False)
|
||||
if inspect_content or hash_content:
|
||||
source_path = self._get_backup_file_from_id(cleaned_metadata["file_id"])
|
||||
if source_path and contained_regular_file(
|
||||
self.target_path, source_path
|
||||
):
|
||||
if inspect_content:
|
||||
enrich_coruna_record(cleaned_metadata, source_path)
|
||||
if hash_content:
|
||||
cleaned_metadata["sha256"] = get_sha256_from_file_path(
|
||||
source_path
|
||||
)
|
||||
self.results.append(cleaned_metadata)
|
||||
|
||||
cur.close()
|
||||
|
||||
@@ -7,15 +7,7 @@ import logging
|
||||
import os
|
||||
from typing import Optional
|
||||
|
||||
from mvt.common.utils import convert_unix_to_iso, get_sha256_from_file_path
|
||||
from mvt.ios.coruna import (
|
||||
alert_coruna_record,
|
||||
contained_regular_file,
|
||||
correlate_coruna_records,
|
||||
coruna_path_artifact,
|
||||
enrich_coruna_record,
|
||||
)
|
||||
from mvt.ios.paths import check_ios_path, normalize_ios_path
|
||||
from mvt.common.utils import convert_unix_to_iso
|
||||
from mvt.common.module_types import (
|
||||
ModuleAtomicResult,
|
||||
ModuleSerializedResult,
|
||||
@@ -57,23 +49,14 @@ class Filesystem(IOSExtraction):
|
||||
}
|
||||
|
||||
def check_indicators(self) -> None:
|
||||
correlate_coruna_records(self.results, self.alertstore)
|
||||
if not self.indicators:
|
||||
return
|
||||
|
||||
for result in self.results:
|
||||
if "path" not in result:
|
||||
continue
|
||||
|
||||
alert_coruna_record(result, self.alertstore)
|
||||
if not self.indicators:
|
||||
continue
|
||||
hash_match = self.indicators.check_file_hash(result.get("sha256", ""))
|
||||
if hash_match:
|
||||
self.alertstore.critical(
|
||||
hash_match.message,
|
||||
result.get("modified", ""),
|
||||
result,
|
||||
matched_indicator=hash_match.ioc,
|
||||
)
|
||||
ioc_match = check_ios_path(self.indicators, result["path"])
|
||||
ioc_match = self.indicators.check_file_path(result["path"])
|
||||
if ioc_match:
|
||||
self.alertstore.high(
|
||||
ioc_match.message, "", result, matched_indicator=ioc_match.ioc
|
||||
@@ -83,9 +66,7 @@ class Filesystem(IOSExtraction):
|
||||
if self.module_options.get("fast_mode", None):
|
||||
continue
|
||||
|
||||
ioc_match = self.indicators.check_file_path_process(
|
||||
normalize_ios_path(result["path"])
|
||||
)
|
||||
ioc_match = self.indicators.check_file_path_process(result["path"])
|
||||
if ioc_match:
|
||||
self.alertstore.high(
|
||||
ioc_match.message, "", result, matched_indicator=ioc_match.ioc
|
||||
@@ -100,12 +81,11 @@ class Filesystem(IOSExtraction):
|
||||
for dir_name in dirs:
|
||||
try:
|
||||
dir_path = os.path.join(root, dir_name)
|
||||
result: ModuleAtomicResult = {
|
||||
result = {
|
||||
"path": os.path.relpath(dir_path, self.target_path).replace(
|
||||
os.sep, "/"
|
||||
),
|
||||
"modified": convert_unix_to_iso(os.stat(dir_path).st_mtime),
|
||||
"is_directory": True,
|
||||
}
|
||||
except Exception:
|
||||
continue
|
||||
@@ -121,14 +101,6 @@ class Filesystem(IOSExtraction):
|
||||
),
|
||||
"modified": convert_unix_to_iso(os.stat(file_path).st_mtime),
|
||||
}
|
||||
inspect_content = coruna_path_artifact(result["path"])
|
||||
hash_content = self.module_options.get("check_file_hashes", False)
|
||||
if (inspect_content or hash_content) and contained_regular_file(
|
||||
self.target_path, file_path
|
||||
):
|
||||
enrich_coruna_record(result, file_path)
|
||||
if hash_content:
|
||||
result["sha256"] = get_sha256_from_file_path(file_path)
|
||||
except Exception:
|
||||
continue
|
||||
else:
|
||||
|
||||
@@ -1,47 +0,0 @@
|
||||
"""Compare device paths independently of the analysis host and backup layout."""
|
||||
|
||||
import posixpath
|
||||
from typing import Optional
|
||||
|
||||
from mvt.common.indicators import IndicatorMatch, Indicators
|
||||
|
||||
|
||||
def normalize_ios_path(path: str) -> str:
|
||||
path = posixpath.normpath("/" + path.replace("\\", "/").lstrip("/"))
|
||||
if path == "/var" or path.startswith("/var/"):
|
||||
return "/private" + path
|
||||
if path == "/tmp" or path.startswith("/tmp/"):
|
||||
return "/private" + path
|
||||
return path
|
||||
|
||||
|
||||
def ios_path_variants(path: str) -> tuple[str, ...]:
|
||||
path = normalize_ios_path(path)
|
||||
if path.startswith(("/private/var/", "/private/tmp/")):
|
||||
return path, path[len("/private") :]
|
||||
return (path,)
|
||||
|
||||
|
||||
def backup_device_path(domain: str, relative_path: str) -> Optional[str]:
|
||||
"""Map only domains with a known fixed root; app UUIDs are not in Manifest.db."""
|
||||
roots = {
|
||||
"HomeDomain": "/private/var/mobile",
|
||||
"CameraRollDomain": "/private/var/mobile",
|
||||
"MediaDomain": "/private/var/mobile/Media",
|
||||
"RootDomain": "/private/var/root",
|
||||
"WirelessDomain": "/private/var/wireless",
|
||||
}
|
||||
root = roots.get(domain)
|
||||
relative_path = relative_path.replace("\\", "/")
|
||||
if not root or relative_path.startswith("/") or ".." in relative_path.split("/"):
|
||||
return None
|
||||
return normalize_ios_path(root + "/" + relative_path)
|
||||
|
||||
|
||||
def check_ios_path(indicators: Indicators, path: str) -> Optional[IndicatorMatch]:
|
||||
for candidate in ios_path_variants(path):
|
||||
match = indicators.check_file_path(candidate)
|
||||
if match:
|
||||
return match
|
||||
# Filesystem records and existing indicators may both use relative paths.
|
||||
return indicators.check_file_path(posixpath.normpath(path.replace("\\", "/")))
|
||||
@@ -0,0 +1,261 @@
|
||||
# Mobile Verification Toolkit (MVT)
|
||||
# Copyright (c) 2021-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 sqlite3
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from mvt.android.modules.androidqf.browser_history import (
|
||||
BrowserHistory as AndroidQFBrowserHistory,
|
||||
)
|
||||
from mvt.android.modules.fs.browser_history import BrowserHistory as FSBrowserHistory
|
||||
from mvt.android.cmd_check_fs import CmdAndroidCheckFS
|
||||
|
||||
CHROME_TIME = 13_348_540_800_000_000
|
||||
URL = "https://example.org/path"
|
||||
|
||||
|
||||
def create_history_database(path: Path, *, url: str = URL) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with sqlite3.connect(path) as connection:
|
||||
connection.executescript(
|
||||
"""
|
||||
CREATE TABLE urls (
|
||||
id INTEGER PRIMARY KEY,
|
||||
url TEXT,
|
||||
title TEXT,
|
||||
visit_count INTEGER,
|
||||
typed_count INTEGER
|
||||
);
|
||||
CREATE TABLE visits (
|
||||
id INTEGER PRIMARY KEY,
|
||||
url INTEGER,
|
||||
visit_time INTEGER,
|
||||
from_visit INTEGER,
|
||||
transition INTEGER
|
||||
);
|
||||
"""
|
||||
)
|
||||
connection.execute("INSERT INTO urls VALUES (1, ?, 'Example', 1, 0)", (url,))
|
||||
connection.execute(
|
||||
"INSERT INTO visits VALUES (7, 1, ?, 0, 805306368)",
|
||||
(CHROME_TIME,),
|
||||
)
|
||||
|
||||
|
||||
def manifest(database_paths: list[tuple[str, str]]) -> dict:
|
||||
return {
|
||||
"schema_version": 1,
|
||||
"status": "collected",
|
||||
"databases": [
|
||||
{
|
||||
"browser": "Chrome",
|
||||
"package": package,
|
||||
"profile": "Default",
|
||||
"device_path": f"/data/data/{package}/app_chrome/Default/History",
|
||||
"archive_path": archive_path,
|
||||
"sidecars": [],
|
||||
}
|
||||
for package, archive_path in database_paths
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("use_zip", [False, True])
|
||||
def test_androidqf_browser_history_directory_and_zip(tmp_path, use_zip):
|
||||
source_database = tmp_path / "source" / "History"
|
||||
create_history_database(source_database)
|
||||
archive_path = "browser_history/com.android.chrome/Default/History"
|
||||
manifest_data = manifest([("com.android.chrome", archive_path)])
|
||||
|
||||
module = AndroidQFBrowserHistory()
|
||||
if use_zip:
|
||||
acquisition_path = tmp_path / "acquisition.zip"
|
||||
with zipfile.ZipFile(acquisition_path, "w") as archive:
|
||||
archive.write(source_database, archive_path)
|
||||
archive.writestr("browser_history/manifest.json", json.dumps(manifest_data))
|
||||
with zipfile.ZipFile(acquisition_path) as archive:
|
||||
module.from_zip(archive, archive.namelist())
|
||||
module.run()
|
||||
else:
|
||||
acquisition = tmp_path / "acquisition"
|
||||
database_path = acquisition / archive_path
|
||||
database_path.parent.mkdir(parents=True)
|
||||
database_path.write_bytes(source_database.read_bytes())
|
||||
manifest_path = acquisition / "browser_history" / "manifest.json"
|
||||
manifest_path.write_text(json.dumps(manifest_data))
|
||||
files = [
|
||||
path.relative_to(tmp_path).as_posix()
|
||||
for path in acquisition.rglob("*")
|
||||
if path.is_file()
|
||||
]
|
||||
module.from_dir(str(tmp_path), files)
|
||||
module.run()
|
||||
|
||||
assert len(module.results) == 1
|
||||
assert module.results[0]["url"] == URL
|
||||
assert module.results[0]["browser"] == "Chrome"
|
||||
assert module.results[0]["source_path"].endswith("/History")
|
||||
module.collect_url_results()
|
||||
module.to_timeline()
|
||||
assert module.url_results[0]["url"] == URL
|
||||
assert module.timeline[0]["event"] == "browser_history"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("prefix", ["", "./", "acquisition/", "./acquisition/"])
|
||||
@pytest.mark.parametrize("separator", ["/", "\\"])
|
||||
def test_androidqf_zip_paths_preserve_wal_visits(tmp_path, prefix, separator):
|
||||
database = tmp_path / "History"
|
||||
create_history_database(database)
|
||||
connection = sqlite3.connect(database)
|
||||
connection.execute("PRAGMA journal_mode=WAL")
|
||||
connection.execute("PRAGMA wal_checkpoint(TRUNCATE)")
|
||||
connection.execute(
|
||||
"INSERT INTO urls VALUES (2, ?, 'WAL', 1, 0)", (URL + "/wal",)
|
||||
)
|
||||
connection.execute(
|
||||
"INSERT INTO visits VALUES (8, 2, ?, 0, 805306368)", (CHROME_TIME,)
|
||||
)
|
||||
connection.commit()
|
||||
archive_path = "browser_history/com.android.chrome/Default/History"
|
||||
acquisition = tmp_path / "acquisition.zip"
|
||||
try:
|
||||
with zipfile.ZipFile(acquisition, "w") as archive:
|
||||
for suffix in ("", "-wal", "-shm"):
|
||||
archive.writestr(
|
||||
(prefix + archive_path + suffix).replace("/", separator),
|
||||
Path(f"{database}{suffix}").read_bytes(),
|
||||
)
|
||||
archive.writestr(
|
||||
(prefix + "browser_history/manifest.json").replace("/", separator),
|
||||
json.dumps(manifest([("com.android.chrome", archive_path)])),
|
||||
)
|
||||
module = AndroidQFBrowserHistory()
|
||||
with zipfile.ZipFile(acquisition) as archive:
|
||||
module.from_zip(archive, archive.namelist())
|
||||
module.run()
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
assert {result["url"] for result in module.results} == {URL, URL + "/wal"}
|
||||
|
||||
|
||||
def test_androidqf_browser_history_isolates_malformed_database(tmp_path, caplog):
|
||||
acquisition = tmp_path / "acquisition"
|
||||
good_archive_path = "browser_history/com.android.chrome/Default/History"
|
||||
bad_archive_path = "browser_history/com.brave.browser/Default/History"
|
||||
create_history_database(acquisition / good_archive_path)
|
||||
bad_path = acquisition / bad_archive_path
|
||||
bad_path.parent.mkdir(parents=True)
|
||||
bad_path.write_bytes(b"not sqlite")
|
||||
manifest_path = acquisition / "browser_history" / "manifest.json"
|
||||
manifest_path.write_text(
|
||||
json.dumps(
|
||||
manifest(
|
||||
[
|
||||
("com.brave.browser", bad_archive_path),
|
||||
("com.android.chrome", good_archive_path),
|
||||
]
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
module = AndroidQFBrowserHistory()
|
||||
files = [
|
||||
path.relative_to(tmp_path).as_posix()
|
||||
for path in acquisition.rglob("*")
|
||||
if path.is_file()
|
||||
]
|
||||
module.from_dir(str(tmp_path), files)
|
||||
module.run()
|
||||
|
||||
assert [result["url"] for result in module.results] == [URL]
|
||||
assert "Unable to parse browser history database" in caplog.text
|
||||
|
||||
|
||||
def test_androidqf_browser_history_rejects_unsafe_manifest_path(tmp_path, caplog):
|
||||
acquisition = tmp_path / "acquisition"
|
||||
manifest_path = acquisition / "browser_history" / "manifest.json"
|
||||
manifest_path.parent.mkdir(parents=True)
|
||||
manifest_path.write_text(
|
||||
json.dumps(manifest([("com.android.chrome", "browser_history/../secret")]))
|
||||
)
|
||||
module = AndroidQFBrowserHistory()
|
||||
module.from_dir(str(tmp_path), [manifest_path.relative_to(tmp_path).as_posix()])
|
||||
|
||||
module.run()
|
||||
|
||||
assert module.results == []
|
||||
assert "unsafe browser history archive path" in caplog.text
|
||||
|
||||
|
||||
def test_filesystem_browser_history_reads_wal_only_visit(tmp_path):
|
||||
database_path = tmp_path / "History"
|
||||
connection = sqlite3.connect(database_path)
|
||||
connection.execute("PRAGMA journal_mode=WAL")
|
||||
connection.executescript(
|
||||
"""
|
||||
CREATE TABLE urls (
|
||||
id INTEGER PRIMARY KEY,
|
||||
url TEXT,
|
||||
title TEXT,
|
||||
visit_count INTEGER,
|
||||
typed_count INTEGER
|
||||
);
|
||||
CREATE TABLE visits (
|
||||
id INTEGER PRIMARY KEY,
|
||||
url INTEGER,
|
||||
visit_time INTEGER,
|
||||
from_visit INTEGER,
|
||||
transition INTEGER
|
||||
);
|
||||
"""
|
||||
)
|
||||
connection.commit()
|
||||
connection.execute("PRAGMA wal_checkpoint(TRUNCATE)")
|
||||
connection.execute("INSERT INTO urls VALUES (1, ?, 'WAL', 1, 0)", (URL,))
|
||||
connection.execute(
|
||||
"INSERT INTO visits VALUES (8, 1, ?, 0, 805306368)", (CHROME_TIME,)
|
||||
)
|
||||
connection.commit()
|
||||
try:
|
||||
assert Path(f"{database_path}-wal").stat().st_size > 0
|
||||
module = FSBrowserHistory(target_path=str(database_path))
|
||||
module.run()
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
assert [result["url"] for result in module.results] == [URL]
|
||||
assert module.results[0]["browser"] == "Chromium"
|
||||
|
||||
|
||||
def test_android_check_fs_finds_all_supported_browser_paths(tmp_path):
|
||||
for index, relative_path in enumerate(
|
||||
(
|
||||
"data/data/com.android.chrome/app_chrome/Default/History",
|
||||
"data/data/com.brave.browser/app_chrome/Default/History",
|
||||
"data/data/com.microsoft.emmx/app_chrome/Default/History",
|
||||
"data/data/com.sec.android.app.sbrowser/app_sbrowser/Default/History",
|
||||
)
|
||||
):
|
||||
create_history_database(
|
||||
tmp_path / relative_path, url=f"https://example.org/{index}"
|
||||
)
|
||||
|
||||
command = CmdAndroidCheckFS(target_path=str(tmp_path), module_name="BrowserHistory")
|
||||
command.run()
|
||||
|
||||
assert len(command.executed) == 1
|
||||
assert {result["package"] for result in command.executed[0].results} == {
|
||||
"com.android.chrome",
|
||||
"com.brave.browser",
|
||||
"com.microsoft.emmx",
|
||||
"com.sec.android.app.sbrowser",
|
||||
}
|
||||
assert len(command.url_results) == 4
|
||||
assert len(command.timeline) == 4
|
||||
@@ -1,284 +0,0 @@
|
||||
import hashlib
|
||||
import json
|
||||
import plistlib
|
||||
import sqlite3
|
||||
|
||||
import pytest
|
||||
|
||||
from mvt.common.alerts import AlertLevel, AlertStore
|
||||
from mvt.common.indicators import Indicators
|
||||
from mvt.ios.coruna import (
|
||||
MAX_ARTIFACT_SIZE,
|
||||
alert_coruna_record,
|
||||
contained_regular_file,
|
||||
correlate_coruna_records,
|
||||
coruna_path_artifact,
|
||||
inspect_coruna_artifact,
|
||||
)
|
||||
from mvt.ios.modules.backup.manifest import Manifest
|
||||
from mvt.ios.modules.fs.filesystem import Filesystem
|
||||
from mvt.ios.paths import backup_device_path, check_ios_path
|
||||
|
||||
PHOTO = "/private/var/mobile/Library/Preferences/com.apple.0123456789abcdef.plist"
|
||||
CONTAINER = "/private/var/mobile/Containers/Data/Application/12345678-1234-1234-1234-123456789abc"
|
||||
PHOTO_DATA = {
|
||||
"LastProcessedFilePath": "/var/mobile/Media/DCIM/100APPLE/IMG_0001.PNG",
|
||||
"LastProcessedTimestamp": 1762537021.5,
|
||||
}
|
||||
CACHE_DATA = {
|
||||
"exp": "plasma",
|
||||
"ecid": "test-ecid",
|
||||
"gci": "test-gci",
|
||||
"unique-id": "test-id",
|
||||
}
|
||||
|
||||
|
||||
def indicators(**values):
|
||||
result = Indicators()
|
||||
result.ioc_collections.append(
|
||||
{"name": "test", "stix2_file_name": "test.stix2", **values}
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"path", [PHOTO, PHOTO[1:], PHOTO.replace("/private", ""), PHOTO.replace("/", "\\")]
|
||||
)
|
||||
def test_photo_state_aliases(path):
|
||||
assert coruna_path_artifact(path) == {"kind": "photo-state"}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"path",
|
||||
[
|
||||
PHOTO + ".bak",
|
||||
PHOTO.replace("abcdef", "abcdeg"),
|
||||
PHOTO.replace("abcdef", "abcde"),
|
||||
PHOTO.replace("/mobile/", "/root/"),
|
||||
"/tmp/" + PHOTO.rsplit("/", 1)[-1],
|
||||
CONTAINER + "/Library/.devcache",
|
||||
"/Documents/.devcache",
|
||||
CONTAINER.replace("/Data/", "/Bundle/") + "/Documents/.devcache",
|
||||
],
|
||||
)
|
||||
def test_artifact_paths_are_scoped(path):
|
||||
assert coruna_path_artifact(path) is None
|
||||
|
||||
|
||||
def test_backup_domain_scoping():
|
||||
relative = "Library/Preferences/" + PHOTO.rsplit("/", 1)[-1]
|
||||
assert coruna_path_artifact(relative, "HomeDomain")
|
||||
assert coruna_path_artifact(relative, "AppDomain-com.example.app") is None
|
||||
assert coruna_path_artifact("Documents/.devcache", "AppDomain-com.example.app")
|
||||
assert coruna_path_artifact("Documents/.devcache", "HomeDomain") is None
|
||||
assert (
|
||||
coruna_path_artifact("Documents/.devcache.bak", "AppDomain-com.example.app")
|
||||
is None
|
||||
)
|
||||
assert (
|
||||
backup_device_path("MediaDomain", "PostLogs.txt")
|
||||
== "/private/var/mobile/Media/PostLogs.txt"
|
||||
)
|
||||
assert (
|
||||
backup_device_path("CameraRollDomain", "Media/DCIM/image.png")
|
||||
== "/private/var/mobile/Media/DCIM/image.png"
|
||||
)
|
||||
assert backup_device_path("HomeDomain", "../root/file") is None
|
||||
assert backup_device_path("AppDomain-com.example.app", "Documents/file") is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"serializer",
|
||||
[
|
||||
plistlib.dumps,
|
||||
lambda d: plistlib.dumps(d, fmt=plistlib.FMT_BINARY),
|
||||
lambda d: json.dumps(d).encode(),
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
"kind,data", [("photo-state", PHOTO_DATA), ("device-cache", CACHE_DATA)]
|
||||
)
|
||||
def test_content_formats(tmp_path, serializer, kind, data):
|
||||
path = tmp_path / "artifact"
|
||||
path.write_bytes(serializer(data))
|
||||
result = inspect_coruna_artifact(str(path), {"kind": kind})
|
||||
assert result["content_status"] == "matched"
|
||||
assert "test-ecid" not in json.dumps(result)
|
||||
assert "IMG_0001.PNG" not in json.dumps(result)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"data",
|
||||
[
|
||||
{},
|
||||
[],
|
||||
{**PHOTO_DATA, "LastProcessedFilePath": "/tmp/photo"},
|
||||
{**PHOTO_DATA, "LastProcessedTimestamp": True},
|
||||
{**PHOTO_DATA, "LastProcessedTimestamp": float("inf")},
|
||||
{"lastProcessedTimestamp": 123},
|
||||
],
|
||||
)
|
||||
def test_photo_content_requires_full_structure(tmp_path, data):
|
||||
path = tmp_path / "artifact"
|
||||
path.write_text(json.dumps(data))
|
||||
assert (
|
||||
inspect_coruna_artifact(str(path), {"kind": "photo-state"})["content_status"]
|
||||
== "unmatched"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"raw",
|
||||
[
|
||||
b"not a plist",
|
||||
b"<?xml version='1.0'?><plist><broken",
|
||||
b"x" * (MAX_ARTIFACT_SIZE + 1),
|
||||
],
|
||||
ids=["invalid", "malformed-xml", "oversized"],
|
||||
)
|
||||
def test_malformed_or_oversized_content(tmp_path, raw):
|
||||
path = tmp_path / "artifact"
|
||||
path.write_bytes(raw)
|
||||
assert (
|
||||
inspect_coruna_artifact(str(path), {"kind": "device-cache"})["content_status"]
|
||||
!= "matched"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"data",
|
||||
[{**CACHE_DATA, "exp": "other"}, {"exp": "plasma"}, {**CACHE_DATA, "ecid": ""}],
|
||||
)
|
||||
def test_device_cache_content_requires_identifiers(tmp_path, data):
|
||||
path = tmp_path / "artifact"
|
||||
path.write_text(json.dumps(data))
|
||||
assert (
|
||||
inspect_coruna_artifact(str(path), {"kind": "device-cache"})["content_status"]
|
||||
== "unmatched"
|
||||
)
|
||||
|
||||
|
||||
def test_correlation_requires_matching_identifiers_in_distinct_containers(tmp_path):
|
||||
path = tmp_path / "cache"
|
||||
path.write_text(json.dumps(CACHE_DATA))
|
||||
a = inspect_coruna_artifact(str(path), {"kind": "device-cache", "container": "one"})
|
||||
store = AlertStore()
|
||||
correlate_coruna_records([{"coruna_artifact": a}, {"coruna_artifact": a}], store)
|
||||
assert not store.alerts
|
||||
correlate_coruna_records(
|
||||
[{"coruna_artifact": a}, {"coruna_artifact": {**a, "container": "two"}}], store
|
||||
)
|
||||
assert len(store.alerts) == 1
|
||||
assert store.alerts[0].event["containers"] == ["one", "two"]
|
||||
assert store.alerts[0].level == AlertLevel.HIGH
|
||||
|
||||
|
||||
def test_path_only_alert_is_medium():
|
||||
store = AlertStore()
|
||||
alert_coruna_record({"path": PHOTO, "isodate": "2026-09-15"}, store)
|
||||
assert store.alerts[0].level == AlertLevel.MEDIUM
|
||||
assert store.alerts[0].event_time == "2026-09-15"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"ioc,path,expected",
|
||||
[
|
||||
("/private/var/tmp/.ghost_upper", "/var/tmp/.ghost_upper/manifest.json", True),
|
||||
("/var/tmp/ioslab_ctl", "private\\var\\tmp\\ioslab_ctl", True),
|
||||
("/var/tmp/ioslab_ctl", "/var/tmp/ioslab_ctl2", False),
|
||||
("/private/var/tmp/.ghost_upper/", "/var/tmp/.ghost_upper-copy/file", False),
|
||||
("matched/directory", "matched/directory/file", True),
|
||||
("matched/directory", "matched\\directory\\file", True),
|
||||
("matched/directory", "matched/directory-copy/file", False),
|
||||
("matched/directory", "matched/directory/../other/file", False),
|
||||
],
|
||||
)
|
||||
def test_ios_path_iocs_respect_aliases_and_boundaries(ioc, path, expected):
|
||||
assert bool(check_ios_path(indicators(file_paths=[ioc]), path)) is expected
|
||||
|
||||
|
||||
def test_filesystem_reads_candidate_and_hashes_when_requested(tmp_path):
|
||||
path = tmp_path / PHOTO.lstrip("/")
|
||||
path.parent.mkdir(parents=True)
|
||||
path.write_bytes(plistlib.dumps(PHOTO_DATA))
|
||||
digest = hashlib.sha256(path.read_bytes()).hexdigest()
|
||||
module = Filesystem(
|
||||
target_path=str(tmp_path), module_options={"check_file_hashes": True}
|
||||
)
|
||||
module.indicators = indicators(files_sha256=[digest])
|
||||
module.run()
|
||||
module.check_indicators()
|
||||
assert sorted(a.level.value for a in module.alertstore.alerts) == [30, 40]
|
||||
record = next(r for r in module.results if "sha256" in r)
|
||||
replay = Filesystem(results=[record])
|
||||
replay.indicators = module.indicators
|
||||
replay.check_indicators()
|
||||
assert sorted(a.level.value for a in replay.alertstore.alerts) == [30, 40]
|
||||
|
||||
|
||||
def test_filesystem_does_not_hash_by_default(tmp_path):
|
||||
(tmp_path / "file").write_bytes(b"example")
|
||||
module = Filesystem(target_path=str(tmp_path))
|
||||
module.run()
|
||||
assert all("sha256" not in r for r in module.results)
|
||||
|
||||
|
||||
def test_backup_reads_coruna_candidate_and_maps_path_ioc(tmp_path):
|
||||
file_id = "a" * 40
|
||||
path = tmp_path / "aa" / file_id
|
||||
path.parent.mkdir()
|
||||
path.write_bytes(plistlib.dumps(PHOTO_DATA))
|
||||
with sqlite3.connect(tmp_path / "Manifest.db") as db:
|
||||
db.execute(
|
||||
"CREATE TABLE Files(fileID TEXT, domain TEXT, relativePath TEXT, flags INTEGER, file BLOB)"
|
||||
)
|
||||
db.execute(
|
||||
"INSERT INTO Files VALUES(?,?,?,?,?)",
|
||||
(
|
||||
file_id,
|
||||
"HomeDomain",
|
||||
"Library/Preferences/" + PHOTO.rsplit("/", 1)[-1],
|
||||
1,
|
||||
None,
|
||||
),
|
||||
)
|
||||
db.execute(
|
||||
"INSERT INTO Files VALUES(?,?,?,?,?)",
|
||||
("b" * 40, "MediaDomain", "PostLogs.txt", 1, None),
|
||||
)
|
||||
db.close()
|
||||
module = Manifest(
|
||||
target_path=str(tmp_path), module_options={"check_file_hashes": True}
|
||||
)
|
||||
module.indicators = indicators(
|
||||
file_paths=["/var/mobile/Media/PostLogs.txt"],
|
||||
files_sha256=[hashlib.sha256(path.read_bytes()).hexdigest()],
|
||||
)
|
||||
module.run()
|
||||
module.check_indicators()
|
||||
assert sorted(a.level.value for a in module.alertstore.alerts) == [30, 30, 40]
|
||||
|
||||
|
||||
def test_acquisition_boundary(tmp_path):
|
||||
root = tmp_path / "root"
|
||||
root.mkdir()
|
||||
outside = tmp_path / "outside"
|
||||
outside.write_bytes(b"not evidence")
|
||||
assert not contained_regular_file(str(root), str(outside))
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"metadata", [{"flags": 2}, {"is_directory": True}, {"mode": 16877}]
|
||||
)
|
||||
def test_directories_are_not_coruna_file_artifacts(metadata):
|
||||
store = AlertStore()
|
||||
alert_coruna_record({"path": PHOTO, **metadata}, store)
|
||||
assert not store.alerts
|
||||
|
||||
|
||||
def test_hash_option_reaches_ios_modules():
|
||||
from mvt.ios.cmd_check_backup import CmdIOSCheckBackup
|
||||
from mvt.ios.cmd_check_fs import CmdIOSCheckFS
|
||||
|
||||
for command in (CmdIOSCheckBackup, CmdIOSCheckFS):
|
||||
assert command(hashes=True).module_options["check_file_hashes"]
|
||||
Reference in new issue
Block a user