Fix AndroidQF artifact discovery in root-level ZIP archives

Bound the ADB manager state by matching braces

src/mvt/android/artifacts/dumpsys_adb.py

@@ -29,11 +29,21 @@ 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:
                 # If the current line is less indented than the previous one, back out
+                if len(stack) <= 1:
+                    # Dedenting below the outermost level means this is not the
+                    # well-formed block the parser expects. Stop here rather than
+                    # raise IndexError out of the module on the next line.
+                    self.log.error(
+                        "Unexpected indentation in ADB manager state, "
+                        "stopping the parse of this section"
+                    )
+                    break
                 stack.pop()
                 cur_indent = indent
             else:
@@ -63,6 +73,12 @@ class DumpsysADBArtifact(AndroidArtifact):
                 current_dict = stack[-1]
 
             if key == "}":
+                if len(stack) <= 1:
+                    self.log.error(
+                        "Unbalanced closing brace in ADB manager state, "
+                        "stopping the parse of this section"
+                    )
+                    break
                 stack.pop()
                 continue
 
@@ -161,6 +177,26 @@ class DumpsysADBArtifact(AndroidArtifact):
                     f"'{user_key['fingerprint']}'"
                 )
 
+    @staticmethod
+    def _find_state_end(content: bytes, open_brace: int) -> int:
+        """Find the unindented line closing the ADB manager state, or -1.
+
+        Braces inside key comments or embedded keystore data are values, while
+        nested structural closing braces are indented.
+        """
+        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:
         """
         Parse the Dumpsys ADB section
@@ -180,12 +216,16 @@ class DumpsysADBArtifact(AndroidArtifact):
             self.log.error("Unable to find ADB manager state in dumpsys output")
             return
 
-        end_of_json = content.rfind(b"}\n")
-        if end_of_json == -1 or end_of_json <= start_of_json:
-            self.log.error("Unable to find complete ADB manager state in dumpsys output")
+        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"
+            )
             return
 
-        json_content = content[start_of_json + 2 : end_of_json - 2].rstrip()
+        # The brace that opens the state and the one that closes it are not part
+        # of the indented body.
+        json_content = content[start_of_json + 2 : end_of_json].rstrip()
 
         parsed = self.indented_dump_parser(json_content)
         if parsed.get("debugging_manager") is None:

src/mvt/android/artifacts/dumpsys_battery_daily.py

@@ -73,8 +73,12 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact):
                 continue
 
             line = line.strip().replace("Update ", "")
-            package_name, vers = line.split(" ", 1)
-            vers_raw = vers.split("=", 1)[1]
+            # A truncated or vendor-specific line must not abort the parse and
+            # lose every record after it.
+            package_name, _, vers = line.partition(" ")
+            vers_raw = vers.partition("=")[2]
+            if not package_name or not vers_raw:
+                continue
             try:
                 version_code: int | str = int(vers_raw)
             except ValueError:

src/mvt/android/modules/androidqf/base.py

@@ -46,8 +46,22 @@ class AndroidQFModule(MVTModule):
         self.archive = archive
         self.files = files
 
-    def _get_files_by_pattern(self, pattern: str):
-        return fnmatch.filter(self.files, pattern)
+    def _get_files_by_pattern(self, pattern: str) -> List[str]:
+        pattern = pattern.replace("\\", "/")
+        patterns = [pattern]
+        # ZIPs may store artifacts at their root, without the acquisition
+        # directory prefix present in paths collected by from_dir().
+        if pattern.startswith("*/"):
+            patterns.append(pattern[2:])
+
+        return [
+            file_path
+            for file_path in self.files
+            if any(
+                fnmatch.fnmatch(file_path.replace("\\", "/"), candidate)
+                for candidate in patterns
+            )
+        ]
 
     def _get_device_timezone(self):
         """

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:

tests/android/test_artifact_dumpsys_adb.py

@@ -130,6 +130,101 @@ 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"
+        b"{\n"
+        b"  debugging_manager={\n"
+        b"    connected_to_adb=true\n"
+        b"    user_keys=QUJDRA== host@example\n"
+        b"  }\n"
+        b"}\n"
+        b"--------- 0.5s was the duration of 'dumpsys adb'\n"
+    )
+
+    def test_a_later_dumpsys_section_does_not_extend_the_adb_state(self):
+        # A bug report holds many sections. Looking for the last closing brace
+        # in the whole output pulled a later section into this one, which threw
+        # 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"
+            b"  debugging_manager={\n"
+            b"    connected_to_adb=false\n"
+            b"    user_keys=RVZJTA== attacker@host\n"
+            b"  }\n"
+            b"}\n"
+        )
+
+        assert len(da_adb.results) == 1
+        assert [key["user"] for key in da_adb.results[0]["user_keys"]] == [
+            "host@example"
+        ]
+        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(
+            b"ADB MANAGER STATE (dumpsys adb):\n"
+            b"{\n"
+            b"  debugging_manager={\n"
+            b"    connected_to_adb=true\n"
+            b"  }\n"
+            b"  }\n"
+            b"}\n"
+        )
+
+        # No exception, and whatever was read before the bad line is kept.
+        assert len(da_adb.results) == 1
+        assert da_adb.results[0]["connected_to_adb"] is True
+
 
 class TestDumpsysADBStateAlerts:
     def test_no_androidqf_context_preserves_existing_behavior(self):

tests/android/test_artifact_dumpsys_battery_daily.py

@@ -144,3 +144,18 @@ class TestDumpsysBatteryDailyArtifact:
             "Detected uninstall of package com.example.app (vers 0)"
         )
         assert uninstall_alert.event_time == "2026-01-10"
+
+    def test_malformed_update_line_does_not_lose_later_records(self):
+        # A truncated "Update" line, or one from a vendor that omits "vers=",
+        # used to raise out of parse() and lose every record after it.
+        dba = DumpsysBatteryDailyArtifact()
+        dba.parse(
+            "  Daily from 2021-05-10-08-00-00 to 2021-05-11-08-00-00:\n"
+            "    Update com.first vers=1\n"
+            "    Update com.truncated\n"
+            "    Update com.no.equals vers 2\n"
+            "    Update com.last vers=3\n"
+        )
+
+        assert [r["package_name"] for r in dba.results] == ["com.first", "com.last"]
+        assert [r["version_code"] for r in dba.results] == [1, 3]

tests/android_androidqf/test_base.py

@@ -0,0 +1,60 @@
+# 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 zipfile
+
+import pytest
+
+from mvt.android.modules.androidqf.base import AndroidQFModule
+
+
+@pytest.mark.parametrize("prefix", ["", "./", "acquisition/", "outer/acquisition/"])
+@pytest.mark.parametrize(
+    "pattern, filename",
+    [
+        ("*/packages.json", "packages.json"),
+        ("*/getprop.txt", "getprop.txt"),
+        ("*/ps.txt", "ps.txt"),
+        ("*/settings_*.txt", "settings_global.txt"),
+        ("*/files.json", "files.json"),
+        ("*/logs/*", "logs/logcat.txt"),
+        ("*/mounts.json", "mounts.json"),
+        ("*/root_binaries.json", "root_binaries.json"),
+    ],
+)
+def test_artifact_lookup_preserves_archive_names(tmp_path, prefix, pattern, filename):
+    member = prefix + filename
+    with zipfile.ZipFile(tmp_path / "acquisition.zip", "w") as archive:
+        archive.writestr(member, b"artifact")
+        archive.writestr(prefix + "unrelated.txt", b"unrelated")
+        module = AndroidQFModule()
+        module.from_zip(archive, archive.namelist())
+
+        assert module._get_files_by_pattern(pattern) == [member]
+        assert module._get_file_content(member) == b"artifact"
+
+
+def test_artifact_lookup_preserves_windows_paths():
+    module = AndroidQFModule()
+    module.from_dir("parent", [r"acquisition\logs\logcat.txt"])
+
+    assert module._get_files_by_pattern("*/logs/*") == [r"acquisition\logs\logcat.txt"]
+
+
+def test_artifact_lookup_preserves_order_without_duplicates():
+    module = AndroidQFModule()
+    module.files = [
+        "acquisition/settings_global.txt",
+        "settings_secure.txt",
+        "./settings_system.txt",
+        "not_settings_global.txt",
+        "settings_global.txt.bak",
+    ]
+
+    assert module._get_files_by_pattern("*/settings_*.txt") == module.files[:3]
+    assert module._get_files_by_pattern("settings_secure.txt") == [
+        "settings_secure.txt"
+    ]
+    assert module._get_files_by_pattern("*/missing.txt") == []

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,

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):

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
 

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)

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)

tests/test_check_android_androidqf.py

@@ -10,6 +10,7 @@ import shutil
 import tempfile
 import zipfile
 
+import pytest
 from click.testing import CliRunner
 
 from mvt.android.cli import check_androidqf
@@ -23,6 +24,44 @@ from .utils import get_artifact_folder
 TEST_BACKUP_PASSWORD = "123456"
 
 
+@pytest.mark.parametrize(
+    "layout", ["directory", "", "./", "acquisition/", "outer/acquisition/"]
+)
+def test_androidqf_property_detection_across_layouts(
+    tmp_path, indicators_factory, monkeypatch, layout
+):
+    monkeypatch.setattr(settings, "NETWORK_ACCESS_ALLOWED", False)
+    content = (
+        "[persist.sys.timezone]: [Europe/Berlin]\n"
+        "[test.suspicious.property]: [enabled]\n"
+    )
+    if layout == "directory":
+        target = tmp_path / "acquisition"
+        target.mkdir()
+        (target / "getprop.txt").write_text(content)
+    else:
+        target = tmp_path / "acquisition.zip"
+        with zipfile.ZipFile(target, "w") as archive:
+            archive.writestr(layout + "getprop.txt", content)
+
+    command = CmdAndroidCheckAndroidQF(
+        target_path=str(target),
+        module_name="AQFGetProp",
+        iocs=indicators_factory(android_property_names=["test.suspicious.property"]),
+    )
+    command.run()
+
+    assert len(command.executed) == 1
+    module = command.executed[0]
+    assert module.results == [
+        {"name": "persist.sys.timezone", "value": "Europe/Berlin"},
+        {"name": "test.suspicious.property", "value": "enabled"},
+    ]
+    assert module._get_device_timezone() == "Europe/Berlin"
+    assert len(command.alertstore.alerts) == 1
+    assert command.alertstore.alerts[0].event == module.results[1]
+
+
 class TestCheckAndroidqfCommand:
     def test_log_timestamps_module_is_registered(self):
         assert AQFLogTimestamps in ANDROIDQF_MODULES
@@ -85,7 +124,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()

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()
 

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"])