diff --git a/src/mvt/android/artifacts/dumpsys_adb.py b/src/mvt/android/artifacts/dumpsys_adb.py index f3c76461..c57cf8d6 100644 --- a/src/mvt/android/artifacts/dumpsys_adb.py +++ b/src/mvt/android/artifacts/dumpsys_adb.py @@ -29,7 +29,8 @@ class DumpsysADBArtifact(AndroidArtifact): stack = [res] cur_indent = 0 in_multiline = False - for line in dump_data.strip(b"\n").split(b"\n"): + for line in dump_data.strip(b"\r\n").split(b"\n"): + line = line.removesuffix(b"\r") # Track the level of indentation indent = len(line) - len(line.lstrip()) if indent < cur_indent: @@ -178,22 +179,22 @@ class DumpsysADBArtifact(AndroidArtifact): @staticmethod def _find_state_end(content: bytes, open_brace: int) -> int: - """Index of the brace closing the one at ``open_brace``, or -1. + """Find the unindented line closing the ADB manager state, or -1. - The end of the ADB manager state is found by matching braces rather than - by looking for the last one in the output: a bug report holds many - dumpsys sections, and a brace in a later one would extend this section - past its end. + Braces inside key comments or embedded keystore data are values, while + nested structural closing braces are indented. """ - depth = 0 - for index in range(open_brace, len(content)): - char = content[index : index + 1] - if char == b"{": - depth += 1 - elif char == b"}": - depth -= 1 - if depth == 0: - return index + line_start = open_brace + while line_start < len(content): + line_end = content.find(b"\n", line_start) + if line_end == -1: + line_end = len(content) + line = content[line_start:line_end].removesuffix(b"\r") + if line_start > open_brace and line == b"}": + return line_start + if line.startswith((b"---------", b"DUMP OF SERVICE ")): + break + line_start = line_end + 1 return -1 def parse(self, content: bytes) -> None: @@ -217,7 +218,9 @@ class DumpsysADBArtifact(AndroidArtifact): end_of_json = self._find_state_end(content, start_of_json + 1) if end_of_json == -1: - self.log.error("Unable to find complete ADB manager state in dumpsys output") + self.log.error( + "Unable to find complete ADB manager state in dumpsys output" + ) return # The brace that opens the state and the one that closes it are not part diff --git a/src/mvt/ios/modules/fs/filesystem.py b/src/mvt/ios/modules/fs/filesystem.py index 563d0394..b3cc1134 100644 --- a/src/mvt/ios/modules/fs/filesystem.py +++ b/src/mvt/ios/modules/fs/filesystem.py @@ -82,7 +82,9 @@ class Filesystem(IOSExtraction): try: dir_path = os.path.join(root, dir_name) result = { - "path": os.path.relpath(dir_path, self.target_path), + "path": os.path.relpath(dir_path, self.target_path).replace( + os.sep, "/" + ), "modified": convert_unix_to_iso(os.stat(dir_path).st_mtime), } except Exception: @@ -94,7 +96,9 @@ class Filesystem(IOSExtraction): try: file_path = os.path.join(root, file_name) result = { - "path": os.path.relpath(file_path, self.target_path), + "path": os.path.relpath(file_path, self.target_path).replace( + os.sep, "/" + ), "modified": convert_unix_to_iso(os.stat(file_path).st_mtime), } except Exception: diff --git a/tests/android/test_artifact_dumpsys_adb.py b/tests/android/test_artifact_dumpsys_adb.py index 232da076..0e5b59f4 100644 --- a/tests/android/test_artifact_dumpsys_adb.py +++ b/tests/android/test_artifact_dumpsys_adb.py @@ -130,7 +130,39 @@ class TestDumpsysADBArtifact: assert key_store_entry["fingerprint"] == expected_fingerprint assert key_store_entry["last_connected"] == "1628501829898" + def test_parsing_adb_xml_with_crlf_line_endings(self): + da_adb = DumpsysADBArtifact() + file = get_artifact("android_data/dumpsys_adb_xml.txt") + with open(file, "rb") as f: + data = f.read().replace(b"\r\n", b"\n").replace(b"\n", b"\r\n") + da_adb.parse(data) + + assert len(da_adb.results) == 1 + assert da_adb.results[0]["user_keys"][0]["user"] == "user@laptop" + assert da_adb.results[0]["keystore"][0]["last_connected"] == "1628501829898" + + def test_parsing_adb_wifi_with_mixed_line_endings(self): + da_adb = DumpsysADBArtifact() + data = ( + b"ADB MANAGER STATE (dumpsys adb):\n" + b"{\n" + b" debugging_manager={\n" + b" connected_to_adb=true\n" + b" user_keys=QUJDRA== host@example\n" + b" adb_wifi={\n" + b" enabled=false\n" + b" }\n" + b" }\r\n" + b"}\n" + b"--------- duration\n" + ) + + da_adb.parse(data) + + assert len(da_adb.results) == 1 + assert da_adb.results[0]["user_keys"][0]["user"] == "host@example" + assert da_adb.results[0]["adb_wifi"]["enabled"] == b"false" ADB_STATE = ( b"ADB MANAGER STATE (dumpsys adb):\n" @@ -149,8 +181,7 @@ class TestDumpsysADBArtifact: # IndexError out of parse() and lost the ADB records entirely. da_adb = DumpsysADBArtifact() da_adb.parse( - self.ADB_STATE - + b"DUMP OF SERVICE other:\n" + self.ADB_STATE + b"DUMP OF SERVICE other:\n" b" debugging_manager={\n" b" connected_to_adb=false\n" b" user_keys=RVZJTA== attacker@host\n" @@ -164,6 +195,20 @@ class TestDumpsysADBArtifact: ] assert da_adb.results[0]["connected_to_adb"] is True + def test_braces_in_key_comment_are_not_state_delimiters(self): + for user in (b"host{example", b"host}}example"): + da_adb = DumpsysADBArtifact() + da_adb.parse( + b"ADB MANAGER STATE (dumpsys adb):\n" + b"{\n" + b" debugging_manager={\n" + b" connected_to_adb=true\n" + b" user_keys=QUJDRA== " + user + b"\n }\n}\n" + ) + + assert len(da_adb.results) == 1 + assert da_adb.results[0]["user_keys"][0]["user"] == user.decode() + def test_unbalanced_state_is_reported_rather_than_raising(self): da_adb = DumpsysADBArtifact() da_adb.parse( diff --git a/tests/common/test_cli_plugins.py b/tests/common/test_cli_plugins.py index 5abb2c98..0618b8d6 100644 --- a/tests/common/test_cli_plugins.py +++ b/tests/common/test_cli_plugins.py @@ -1,3 +1,4 @@ +import shlex from types import SimpleNamespace import click @@ -110,7 +111,7 @@ def test_load_command_option_supports_folders_and_repeated_paths(tmp_path): def test_loaded_command_participates_in_shell_completion(tmp_path): command_path = _write_command(tmp_path / "hello.py", "hello") group = _make_group() - words = f"group --load-command {command_path} he" + words = f"group --load-command {shlex.quote(str(command_path))} he" result = CliRunner().invoke( group, diff --git a/tests/common/test_cmd_plugins.py b/tests/common/test_cmd_plugins.py index adf93227..3767491a 100644 --- a/tests/common/test_cmd_plugins.py +++ b/tests/common/test_cmd_plugins.py @@ -1,4 +1,5 @@ import json +import re from types import SimpleNamespace import pytest @@ -54,20 +55,23 @@ def _run(command, arguments): return CliRunner().invoke(command, arguments, env={"COLUMNS": "200"}) -def _table_rows(output): - """Return the content of the table rows, without the header and the box.""" +def _table_lines(output): + """Return the cells of each line of the table, header first.""" return [ - [cell.strip() for cell in line.strip().strip("│").split("│")] + [cell.strip() for cell in re.split("[│┃]", line.strip().strip("│┃"))] for line in output.splitlines() - if "│" in line + if "│" in line or "┃" in line ] +def _table_rows(output): + """Return the content of the table rows, without the header and the box.""" + return _table_lines(output)[1:] + + def _table_header(output): - for line in output.splitlines(): - if "┃" in line: - return [cell.strip() for cell in line.strip().strip("┃").split("┃")] - return [] + lines = _table_lines(output) + return lines[0] if lines else [] def _install(monkeypatch, distributions, entry_points): diff --git a/tests/ios_backup/test_decrypt.py b/tests/ios_backup/test_decrypt.py index 25decce5..1e2ce148 100644 --- a/tests/ios_backup/test_decrypt.py +++ b/tests/ios_backup/test_decrypt.py @@ -7,6 +7,7 @@ import logging import threading from pathlib import Path +import pytest from Crypto.Cipher import AES from mvt.ios.decrypt import DecryptBackup, MVTEncryptedBackup @@ -81,7 +82,10 @@ def test_extract_file_by_id_copies_unencrypted_files(mocker, tmp_path): assert output_path.read_bytes() == b"plain content" -def test_process_backup_rejects_unsafe_file_ids_and_destinations(mocker, tmp_path): +@pytest.mark.parametrize("with_symlink", [False, True], ids=["file-id", "symlink"]) +def test_process_backup_rejects_unsafe_file_ids_and_destinations( + mocker, tmp_path, with_symlink +): backup_path = tmp_path / "backup" destination = tmp_path / "destination" outside = tmp_path / "outside" @@ -92,20 +96,27 @@ def test_process_backup_rejects_unsafe_file_ids_and_destinations(mocker, tmp_pat safe_file_id = "ef" + "3" * 38 unsafe_file_id = "../../outside-file" symlink_file_id = "ab" + "4" * 38 - for file_id in (safe_file_id, symlink_file_id): + file_ids = [safe_file_id] + if with_symlink: + file_ids.append(symlink_file_id) + for file_id in file_ids: source_path = backup_path / file_id[:2] / file_id source_path.parent.mkdir(parents=True, exist_ok=True) source_path.write_bytes(b"encrypted") - (destination / "ab").symlink_to(outside, target_is_directory=True) + if with_symlink: + try: + (destination / "ab").symlink_to(outside, target_is_directory=True) + except OSError: + pytest.skip("creating symbolic links is not permitted on this system") cursor = mocker.MagicMock() - cursor.__iter__.return_value = iter( - [ - (safe_file_id, "Domain", "safe", b"plist"), - (unsafe_file_id, "Domain", "unsafe", b"plist"), - (symlink_file_id, "Domain", "symlink", b"plist"), - ] - ) + records = [ + (safe_file_id, "Domain", "safe", b"plist"), + (unsafe_file_id, "Domain", "unsafe", b"plist"), + ] + if with_symlink: + records.append((symlink_file_id, "Domain", "symlink", b"plist")) + cursor.__iter__.return_value = iter(records) cursor_context = mocker.MagicMock() cursor_context.__enter__.return_value = cursor @@ -124,7 +135,8 @@ def test_process_backup_rejects_unsafe_file_ids_and_destinations(mocker, tmp_pat decryptor.process_backup() assert (destination / safe_file_id[:2] / safe_file_id).read_bytes() == b"decrypted" - assert not (outside / symlink_file_id).exists() + if with_symlink: + assert not (outside / symlink_file_id).exists() backup.extract_file_by_id.assert_called_once() assert backup.extract_file_by_id.call_args.kwargs["file_id"] == safe_file_id diff --git a/tests/ios_backup/test_sqlite_handling.py b/tests/ios_backup/test_sqlite_handling.py index f7806401..2774098e 100644 --- a/tests/ios_backup/test_sqlite_handling.py +++ b/tests/ios_backup/test_sqlite_handling.py @@ -9,6 +9,8 @@ import plistlib import shutil import sqlite3 +import pytest + from mvt.ios.modules.base import IOSExtraction from mvt.ios.modules.fs.analytics import Analytics @@ -45,6 +47,9 @@ def test_open_sqlite_reads_wal_without_modifying_evidence(tmp_path): assert not os.path.exists(str(evidence_path) + "-shm") +@pytest.mark.skipif( + shutil.which("sqlite3") is None, reason="the recovery needs the sqlite3 binary" +) def test_recovery_preserves_source_database(tmp_path): database_path = tmp_path / "source.db" conn = sqlite3.connect(database_path) diff --git a/tests/ios_fs/test_filesystem.py b/tests/ios_fs/test_filesystem.py index 636c004a..fe3ecf68 100644 --- a/tests/ios_fs/test_filesystem.py +++ b/tests/ios_fs/test_filesystem.py @@ -3,15 +3,45 @@ # 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 ntpath +from types import SimpleNamespace from mvt.common.indicators import Indicators from mvt.common.module import run_module +from mvt.ios.modules.fs import filesystem from mvt.ios.modules.fs.filesystem import Filesystem from ..utils import get_ios_backup_folder class TestFilesystem: + def test_windows_paths_are_normalized_for_indicators( + self, monkeypatch, indicators_factory + ): + m = Filesystem(target_path=r"C:\dump") + m.indicators = indicators_factory( + file_paths=["matched/directory"], processes=["matched"] + ) + monkeypatch.setattr( + filesystem, + "os", + SimpleNamespace( + sep="\\", + path=ntpath, + walk=lambda _: [(r"C:\dump\matched", ["directory"], ["file.txt"])], + stat=lambda _: SimpleNamespace(st_mtime=0), + ), + ) + + m.run() + m.check_indicators() + + assert {result["path"] for result in m.results} == { + "matched/directory", + "matched/file.txt", + } + assert len(m.alertstore.alerts) == 3 + def test_filesystem(self): m = Filesystem(target_path=get_ios_backup_folder()) run_module(m) diff --git a/tests/test_check_android_androidqf.py b/tests/test_check_android_androidqf.py index 4e6ef984..65973109 100644 --- a/tests/test_check_android_androidqf.py +++ b/tests/test_check_android_androidqf.py @@ -85,7 +85,7 @@ class TestCheckAndroidqfCommand: def test_acquisition_context_falls_back_to_public_key_file(self, tmp_path): data_path = tmp_path / "androidqf" data_path.mkdir() - (data_path / "adb_host_key.pub").write_text("QUJDRA== acquisition@host\n") + (data_path / "adb_host_key.pub").write_bytes(b"QUJDRA== acquisition@host\n") command = CmdAndroidCheckAndroidQF(target_path=str(data_path)) command.init() diff --git a/tests/test_cmd_check_sysdiagnose.py b/tests/test_cmd_check_sysdiagnose.py index b00b648b..702e985f 100644 --- a/tests/test_cmd_check_sysdiagnose.py +++ b/tests/test_cmd_check_sysdiagnose.py @@ -116,7 +116,7 @@ def test_archive_is_extracted_once_and_unsafe_members_are_skipped(tmp_path): module = SysdiagnoseExtraction() command.module_init(module) assert module.tar is None - assert module.parent_path == str(extracted_path.parent) + assert Path(module.parent_path) == extracted_path.parent finally: command.finish() diff --git a/tests/test_completion.py b/tests/test_completion.py index 08ad3ce3..d240e492 100644 --- a/tests/test_completion.py +++ b/tests/test_completion.py @@ -43,6 +43,7 @@ class TestCompletionCommand: def test_completion_install_updates_bashrc_once(self, tmp_path, monkeypatch): monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) runner = CliRunner() result = runner.invoke(mvt_cli, ["completion", "bash", "--install"]) @@ -67,6 +68,7 @@ class TestCompletionCommand: self, tmp_path, monkeypatch ): monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) runner = CliRunner() result = runner.invoke(mvt_cli, ["completion", "fish", "--install"])