Add accessibility enabled-state logging and mixed-dump coverage

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/bugreport/dumpsys_accessibility.py

@@ -48,12 +48,17 @@ class DumpsysAccessibility(DumpsysAccessibilityArtifact, BugReportModule):
         )
         self.parse(content)
 
-        listed = unnamed = 0
+        listed = unnamed = enabled_count = 0
         for result in self.results:
             if result.get("component"):
                 listed += 1
+                enabled = result.get("enabled")
+                if enabled is True:
+                    enabled_count += 1
                 self.log.info(
-                    'Found accessibility service "%s"', result.get("component")
+                    'Found accessibility service "%s" (enabled: %s)',
+                    result.get("component"),
+                    "not stated" if enabled is None else enabled,
                 )
                 continue
             # The operator gets this per user as a LOW alert from
@@ -62,8 +67,9 @@ class DumpsysAccessibility(DumpsysAccessibilityArtifact, BugReportModule):
             unnamed += result.get("unnamed_service_count") or 0
 
         self.log.info(
-            "Identified a total of %d accessibility services (%d more stated by "
-            "the dump without a component name)",
+            "Identified a total of %d accessibility services, %d reported enabled "
+            "(%d more stated by the dump without a component name)",
             listed,
+            enabled_count,
             unnamed,
         )

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_accessibility.py

@@ -46,6 +46,23 @@ class TestDumpsysAccessibilityArtifact:
         # "not stated" into "not installed", so it reads None here.
         assert da.results[0]["installed"] is None
 
+    def test_parsing_installed_and_enabled(self):
+        da = DumpsysAccessibilityArtifact()
+        with open(get_artifact("android_data/dumpsys_accessibility_enabled.txt")) as f:
+            da.parse(f.read())
+
+        assert len(da.results) == 5
+        assert all(result["component"] for result in da.results)
+        assert all(result["installed"] is True for result in da.results)
+        enabled = [result for result in da.results if result["enabled"] is True]
+        assert len(enabled) == 1
+        assert enabled[0]["package_name"] == "com.samsung.accessibility"
+        assert (
+            enabled[0]["component"]
+            == "com.samsung.accessibility/.universalswitch.UniversalSwitchService"
+        )
+        assert sum(result["enabled"] is False for result in da.results) == 4
+
     def test_accessibility_service_alert(self):
         da = DumpsysAccessibilityArtifact()
         file = get_artifact("android_data/dumpsys_accessibility_v14_or_later.txt")

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/test_bugreport_dumpsys_accessibility.py

@@ -0,0 +1,73 @@
+# 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 mvt.android.modules.bugreport.dumpsys_accessibility import DumpsysAccessibility
+
+from ..utils import get_artifact
+
+
+def _run(content, monkeypatch, caplog):
+    module = DumpsysAccessibility()
+    dumpstate = f"DUMP OF SERVICE accessibility:\n{content}".encode()
+    monkeypatch.setattr(module, "_get_dumpstate_file", lambda: dumpstate)
+    with caplog.at_level(logging.INFO):
+        module.run()
+
+
+def test_logs_mixed_enabled_states(monkeypatch, caplog):
+    with open(get_artifact("android_data/dumpsys_accessibility_enabled.txt")) as handle:
+        _run(handle.read(), monkeypatch, caplog)
+
+    assert (
+        'Found accessibility service "com.samsung.accessibility/'
+        '.universalswitch.UniversalSwitchService" (enabled: True)' in caplog.messages
+    )
+    assert sum("(enabled: False)" in message for message in caplog.messages) == 4
+    assert (
+        "Identified a total of 5 accessibility services, 1 reported enabled "
+        "(0 more stated by the dump without a component name)" in caplog.messages
+    )
+
+
+def test_logs_enabled_count_alongside_unnamed_count(monkeypatch, caplog):
+    with open(
+        get_artifact("android_data/dumpsys_accessibility_v14_or_later.txt")
+    ) as handle:
+        _run(handle.read(), monkeypatch, caplog)
+
+    assert (
+        "Identified a total of 1 accessibility services, 1 reported enabled "
+        "(1 more stated by the dump without a component name)" in caplog.messages
+    )
+    assert (
+        sum("Found accessibility service" in message for message in caplog.messages)
+        == 1
+    )
+
+
+def test_logs_unknown_enabled_state_without_treating_it_as_disabled(
+    monkeypatch, caplog
+):
+    _run(
+        """User state[attributes:{id=0, installedServiceCount=2}
+  installed services: {
+    0 : com.example/.Service
+  }
+""",
+        monkeypatch,
+        caplog,
+    )
+
+    assert (
+        'Found accessibility service "com.example/.Service" (enabled: not stated)'
+        in caplog.messages
+    )
+    assert not any("(enabled: False)" in message for message in caplog.messages)
+    assert (
+        "Identified a total of 1 accessibility services, 0 reported enabled "
+        "(1 more stated by the dump without a component name)" in caplog.messages
+    )

tests/artifacts/android_data/dumpsys_accessibility_enabled.txt

@@ -0,0 +1,16 @@
+ACCESSIBILITY MANAGER (dumpsys accessibility)
+
+currentUserId=0
+User state[
+     attributes:{id=0, touchExplorationEnabled=false, installedServiceCount=5}
+     installed services: {
+      0 : com.google.android.apps.accessibility.voiceaccess/.JustSpeakService  (A11yTool)
+      1 : com.microsoft.appmanager/com.microsoft.mmx.screenmirroringsrc.accessibility.ScreenMirroringAccessibilityService
+      2 : com.samsung.accessibility/.assistantmenu.serviceframework.AssistantMenuService  (A11yTool)
+      3 : com.samsung.accessibility/.universalswitch.UniversalSwitchService  (A11yTool)
+      4 : com.samsung.android.accessibility.talkback/com.samsung.android.marvin.talkback.TalkBackService  (A11yTool)
+     }
+     Bound services:{}
+     Enabled services:{{com.samsung.accessibility/.universalswitch.UniversalSwitchService}}
+     Binding services:{}
+     Crashed services:{}

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

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

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