Show which accessibility services are enabled

Parse both installed and enabled accessibility services, adding an
"enabled" field to each result. This lets users see at a glance
whether any installed accessibility service is actually active.

Fixes #744

pyproject.toml

@@ -24,8 +24,7 @@ dependencies = [
     "simplejson==3.20.2",
     "packaging==26.0",
     "appdirs==1.4.4",
-    "iphone_backup_decrypt==0.9.0",
-    "pycryptodome>=3.18",
+    "iOSbackup==0.9.925",
     "adb-shell[usb]==0.4.4",
     "libusb1==3.3.1",
     "cryptography==46.0.6",

src/mvt/android/artifacts/dumpsys_accessibility.py

@@ -22,13 +22,13 @@ class DumpsysAccessibilityArtifact(AndroidArtifact):
 
     def parse(self, content: str) -> None:
         """
-        Parse the Dumpsys Accessibility section/
-        Adds results to self.results (List[Dict[str, str]])
+        Parse the Dumpsys Accessibility section.
+        Adds results to self.results (List[Dict[str, Any]])
 
         :param content: content of the accessibility section (string)
         """
 
-        # "Old" syntax
+        # Parse installed services
         in_services = False
         for line in content.splitlines():
             if line.strip().startswith("installed services:"):
@@ -39,7 +39,6 @@ class DumpsysAccessibilityArtifact(AndroidArtifact):
                 continue
 
             if line.strip() == "}":
-                # At end of installed services
                 break
 
             service = line.split(":")[1].strip()
@@ -48,21 +47,66 @@ class DumpsysAccessibilityArtifact(AndroidArtifact):
                 {
                     "package_name": service.split("/")[0],
                     "service": service,
+                    "enabled": False,
                 }
             )
 
-        # "New" syntax - AOSP >= 14 (?)
-        # Looks like:
-        # Enabled services:{{com.azure.authenticator/com.microsoft.brooklyn.module.accessibility.BrooklynAccessibilityService}, {com.agilebits.onepassword/com.agilebits.onepassword.filling.accessibility.FillingAccessibilityService}}
+        # Parse enabled services from both old and new formats.
+        #
+        # Old format (multi-line block):
+        #   enabled services: {
+        #     0 : com.example/.MyService
+        #   }
+        #
+        # New format (single line, AOSP >= 14):
+        #   Enabled services:{{com.example/com.example.MyService}, {com.other/com.other.Svc}}
+        enabled_services = set()
 
+        in_enabled = False
         for line in content.splitlines():
-            if line.strip().startswith("Enabled services:"):
-                matches = re.finditer(r"{([^{]+?)}", line)
+            stripped = line.strip()
 
+            if in_enabled:
+                if stripped == "}":
+                    in_enabled = False
+                    continue
+                service = line.split(":")[1].strip()
+                enabled_services.add(service)
+                continue
+
+            if re.match(r"enabled services:\s*\{\s*$", stripped, re.IGNORECASE):
+                # Old multi-line format: "enabled services: {"
+                in_enabled = True
+                continue
+
+            if re.match(r"enabled services:\s*\{", stripped, re.IGNORECASE):
+                # New single-line format: "Enabled services:{{pkg/svc}, {pkg2/svc2}}"
+                matches = re.finditer(r"\{([^{}]+)\}", stripped)
                 for match in matches:
-                    # Each match is in format: <package_name>/<service>
-                    package_name, _, service = match.group(1).partition("/")
+                    enabled_services.add(match.group(1).strip())
 
-                    self.results.append(
-                        {"package_name": package_name, "service": service}
-                    )
+        # Mark installed services that are enabled.
+        # Installed service names may include trailing annotations like
+        # "(A11yTool)" that are absent from the enabled services list,
+        # so strip annotations before comparing.
+        def _strip_annotation(s: str) -> str:
+            return re.sub(r"\s+\(.*?\)\s*$", "", s)
+
+        installed_stripped = {
+            _strip_annotation(r["service"]): r for r in self.results
+        }
+        for enabled in enabled_services:
+            if enabled in installed_stripped:
+                installed_stripped[enabled]["enabled"] = True
+
+        # Add enabled services not found in the installed list
+        for service in enabled_services:
+            if service not in installed_stripped:
+                package_name, _, _ = service.partition("/")
+                self.results.append(
+                    {
+                        "package_name": package_name,
+                        "service": service,
+                        "enabled": True,
+                    }
+                )

src/mvt/android/modules/bugreport/dumpsys_accessibility.py

@@ -49,9 +49,14 @@ class DumpsysAccessibility(DumpsysAccessibilityArtifact, BugReportModule):
 
         for result in self.results:
             self.log.info(
-                'Found installed accessibility service "%s"', result.get("service")
+                'Found installed accessibility service "%s" (enabled: %s)',
+                result.get("service"),
+                result.get("enabled", False),
             )
 
+        enabled_count = sum(1 for r in self.results if r.get("enabled"))
         self.log.info(
-            "Identified a total of %d accessibility services", len(self.results)
+            "Identified a total of %d accessibility services, %d enabled",
+            len(self.results),
+            enabled_count,
         )

src/mvt/ios/decrypt.py

@@ -6,146 +6,17 @@
 import binascii
 import glob
 import logging
+import multiprocessing
 import os
 import os.path
-import plistlib
 import shutil
 import sqlite3
-import tempfile
 from typing import Optional
 
-from iphone_backup_decrypt import EncryptedBackup
-from iphone_backup_decrypt import google_iphone_dataprotection
+from iOSbackup import iOSbackup
 
 log = logging.getLogger(__name__)
 
-# Import pbkdf2_hmac from the same source iphone_backup_decrypt uses internally,
-# so our key derivation is consistent with theirs.
-try:
-    from fastpbkdf2 import pbkdf2_hmac
-except ImportError:
-    import Crypto.Hash.SHA1
-    import Crypto.Hash.SHA256
-    import Crypto.Protocol.KDF
-
-    _HASH_FNS = {"sha1": Crypto.Hash.SHA1, "sha256": Crypto.Hash.SHA256}
-
-    def pbkdf2_hmac(hash_name, password, salt, iterations, dklen=None):
-        return Crypto.Protocol.KDF.PBKDF2(
-            password, salt, dklen, iterations, hmac_hash_module=_HASH_FNS[hash_name]
-        )
-
-
-class MVTEncryptedBackup(EncryptedBackup):
-    """Extends EncryptedBackup with derived key export/import.
-
-    NOTE: This subclass relies on internal APIs of iphone_backup_decrypt
-    (specifically _read_and_unlock_keybag, _keybag, and the Keybag class
-    internals). Pinned to iphone_backup_decrypt==0.9.0.
-    """
-
-    def __init__(self, *, backup_directory, passphrase=None, derived_key=None):
-        if passphrase:
-            super().__init__(backup_directory=backup_directory, passphrase=passphrase)
-            self._derived_key = None  # Will be set after keybag unlock
-        elif derived_key:
-            self._init_without_passphrase(backup_directory, derived_key)
-        else:
-            raise ValueError("Either passphrase or derived_key must be provided")
-
-    def _init_without_passphrase(self, backup_directory, derived_key):
-        """Replicate parent __init__ state without requiring a passphrase."""
-        self.decrypted = False
-        self._backup_directory = os.path.expandvars(backup_directory)
-        self._passphrase = None
-        self._manifest_plist_path = os.path.join(
-            self._backup_directory, "Manifest.plist"
-        )
-        self._manifest_plist = None
-        self._manifest_db_path = os.path.join(self._backup_directory, "Manifest.db")
-        self._keybag = None
-        self._unlocked = False
-        self._temporary_folder = tempfile.mkdtemp()
-        self._temp_decrypted_manifest_db_path = os.path.join(
-            self._temporary_folder, "Manifest.db"
-        )
-        self._temp_manifest_db_conn = None
-        self._derived_key = derived_key  # 32 raw bytes
-
-    def _read_and_unlock_keybag(self):
-        """Override to capture derived key on password unlock, or use
-        a pre-derived key to skip PBKDF2."""
-        if self._unlocked:
-            return self._unlocked
-
-        with open(self._manifest_plist_path, "rb") as infile:
-            self._manifest_plist = plistlib.load(infile)
-        self._keybag = google_iphone_dataprotection.Keybag(
-            self._manifest_plist["BackupKeyBag"]
-        )
-
-        if self._derived_key:
-            # Skip PBKDF2, unwrap class keys directly with pre-derived key
-            self._unlocked = _unlock_keybag_with_derived_key(
-                self._keybag, self._derived_key
-            )
-        else:
-            # Normal path: full PBKDF2 derivation, capturing the intermediate key
-            self._unlocked, self._derived_key = _unlock_keybag_and_capture_key(
-                self._keybag, self._passphrase
-            )
-            self._passphrase = None
-
-        if not self._unlocked:
-            raise ValueError("Failed to decrypt keys: incorrect passphrase?")
-        return True
-
-    def get_decryption_key(self):
-        """Return derived key as hex string (64 chars / 32 bytes)."""
-        if self._derived_key is None:
-            raise ValueError("No derived key available")
-        return self._derived_key.hex()
-
-
-def _unlock_keybag_with_derived_key(keybag, passphrase_key):
-    """Unlock keybag class keys using a pre-derived passphrase_key,
-    skipping the expensive PBKDF2 rounds."""
-    WRAP_PASSPHRASE = 2
-    for classkey in keybag.classKeys.values():
-        if b"WPKY" not in classkey:
-            continue
-        if classkey[b"WRAP"] & WRAP_PASSPHRASE:
-            k = google_iphone_dataprotection._AESUnwrap(
-                passphrase_key, classkey[b"WPKY"]
-            )
-            if not k:
-                return False
-            classkey[b"KEY"] = k
-    return True
-
-
-def _unlock_keybag_and_capture_key(keybag, passphrase):
-    """Run full PBKDF2 key derivation and AES unwrap, returning
-    (success, passphrase_key) so the derived key can be exported."""
-    passphrase_round1 = pbkdf2_hmac(
-        "sha256", passphrase, keybag.attrs[b"DPSL"], keybag.attrs[b"DPIC"], 32
-    )
-    passphrase_key = pbkdf2_hmac(
-        "sha1", passphrase_round1, keybag.attrs[b"SALT"], keybag.attrs[b"ITER"], 32
-    )
-    WRAP_PASSPHRASE = 2
-    for classkey in keybag.classKeys.values():
-        if b"WPKY" not in classkey:
-            continue
-        if classkey[b"WRAP"] & WRAP_PASSPHRASE:
-            k = google_iphone_dataprotection._AESUnwrap(
-                passphrase_key, classkey[b"WPKY"]
-            )
-            if not k:
-                return False, None
-            classkey[b"KEY"] = k
-    return True, passphrase_key
-
 
 class DecryptBackup:
     """This class provides functions to decrypt an encrypted iTunes backup
@@ -184,27 +55,41 @@ class DecryptBackup:
             log.critical("The backup does not seem encrypted!")
             return False
 
+    def _process_file(
+        self, relative_path: str, domain: str, item, file_id: str, item_folder: str
+    ) -> None:
+        self._backup.getFileDecryptedCopy(
+            manifestEntry=item, targetName=file_id, targetFolder=item_folder
+        )
+        log.info(
+            "Decrypted file %s [%s] to %s/%s",
+            relative_path,
+            domain,
+            item_folder,
+            file_id,
+        )
+
     def process_backup(self) -> None:
         if not os.path.exists(self.dest_path):
             os.makedirs(self.dest_path)
 
         manifest_path = os.path.join(self.dest_path, "Manifest.db")
-        # Extract a decrypted Manifest.db to the destination folder.
-        self._backup.save_manifest_file(output_filename=manifest_path)
+        # We extract a decrypted Manifest.db.
+        self._backup.getManifestDB()
+        # We store it to the destination folder.
+        shutil.copy(self._backup.manifestDB, manifest_path)
+
+        pool = multiprocessing.Pool(multiprocessing.cpu_count())
+
+        for item in self._backup.getBackupFilesList():
+            try:
+                file_id = item["backupFile"]
+                relative_path = item["relativePath"]
+                domain = item["domain"]
 
-        # Iterate over all files in the backup and decrypt them,
-        # preserving the XX/file_id directory structure that downstream
-        # modules expect.
-        with self._backup.manifest_db_cursor() as cur:
-            cur.execute(
-                "SELECT fileID, domain, relativePath, file FROM Files WHERE flags=1"
-            )
-            for file_id, domain, relative_path, file_bplist in cur:
                 # This may be a partial backup. Skip files from the manifest
                 # which do not exist locally.
-                source_file_path = os.path.join(
-                    self.backup_path, file_id[:2], file_id
-                )
+                source_file_path = os.path.join(self.backup_path, file_id[0:2], file_id)
                 if not os.path.exists(source_file_path):
                     log.debug(
                         "Skipping file %s. File not found in encrypted backup directory.",
@@ -212,26 +97,24 @@ class DecryptBackup:
                     )
                     continue
 
-                item_folder = os.path.join(self.dest_path, file_id[:2])
-                os.makedirs(item_folder, exist_ok=True)
+                item_folder = os.path.join(self.dest_path, file_id[0:2])
+                if not os.path.exists(item_folder):
+                    os.makedirs(item_folder)
 
-                try:
-                    decrypted = self._backup._decrypt_inner_file(
-                        file_id=file_id, file_bplist=file_bplist
-                    )
-                    with open(
-                        os.path.join(item_folder, file_id), "wb"
-                    ) as handle:
-                        handle.write(decrypted)
-                    log.info(
-                        "Decrypted file %s [%s] to %s/%s",
-                        relative_path,
-                        domain,
-                        item_folder,
-                        file_id,
-                    )
-                except Exception as exc:
-                    log.error("Failed to decrypt file %s: %s", relative_path, exc)
+                # iOSBackup getFileDecryptedCopy() claims to read a "file"
+                # parameter but the code actually is reading the "manifest" key.
+                # Add manifest plist to both keys to handle this.
+                item["manifest"] = item["file"]
+
+                pool.apply_async(
+                    self._process_file,
+                    args=(relative_path, domain, item, file_id, item_folder),
+                )
+            except Exception as exc:
+                log.error("Failed to decrypt file %s: %s", relative_path, exc)
+
+        pool.close()
+        pool.join()
 
         # Copying over the root plist files as well.
         for file_name in os.listdir(self.backup_path):
@@ -272,23 +155,20 @@ class DecryptBackup:
             return
 
         try:
-            self._backup = MVTEncryptedBackup(
-                backup_directory=self.backup_path,
-                passphrase=password,
+            self._backup = iOSbackup(
+                udid=os.path.basename(self.backup_path),
+                cleartextpassword=password,
+                backuproot=os.path.dirname(self.backup_path),
             )
-            # Eagerly trigger keybag unlock so wrong-password errors
-            # surface here rather than later during process_backup().
-            self._backup.test_decryption()
         except Exception as exc:
-            self._backup = None
             if (
-                isinstance(exc, ValueError)
-                and "passphrase" in str(exc).lower()
+                isinstance(exc, KeyError)
+                and len(exc.args) > 0
+                and exc.args[0] == b"KEY"
             ):
                 log.critical("Failed to decrypt backup. Password is probably wrong.")
             elif (
                 isinstance(exc, FileNotFoundError)
-                and hasattr(exc, "filename")
                 and os.path.basename(exc.filename) == "Manifest.plist"
             ):
                 log.critical(
@@ -331,14 +211,12 @@ class DecryptBackup:
 
         try:
             key_bytes_raw = binascii.unhexlify(key_bytes)
-            self._backup = MVTEncryptedBackup(
-                backup_directory=self.backup_path,
-                derived_key=key_bytes_raw,
+            self._backup = iOSbackup(
+                udid=os.path.basename(self.backup_path),
+                derivedkey=key_bytes_raw,
+                backuproot=os.path.dirname(self.backup_path),
             )
-            # Eagerly trigger keybag unlock so wrong-key errors surface here.
-            self._backup.test_decryption()
         except Exception as exc:
-            self._backup = None
             log.exception(exc)
             log.critical(
                 "Failed to decrypt backup. Did you provide the correct key file?"
@@ -349,7 +227,7 @@ class DecryptBackup:
         if not self._backup:
             return
 
-        self._decryption_key = self._backup.get_decryption_key()
+        self._decryption_key = self._backup.getDecryptionKey()
         log.info(
             'Derived decryption key for backup at path %s is: "%s"',
             self.backup_path,

tests/android/test_artifact_dumpsys_accessibility.py

@@ -25,6 +25,9 @@ class TestDumpsysAccessibilityArtifact:
             da.results[0]["service"]
             == "com.android.settings/com.samsung.android.settings.development.gpuwatch.GPUWatchInterceptor"
         )
+        # All services are installed but none enabled in this fixture
+        for result in da.results:
+            assert result["enabled"] is False
 
     def test_parsing_v14_aosp_format(self):
         da = DumpsysAccessibilityArtifact()
@@ -36,7 +39,32 @@ class TestDumpsysAccessibilityArtifact:
         da.parse(data)
         assert len(da.results) == 1
         assert da.results[0]["package_name"] == "com.malware.accessibility"
-        assert da.results[0]["service"] == "com.malware.service.malwareservice"
+        assert (
+            da.results[0]["service"]
+            == "com.malware.accessibility/com.malware.service.malwareservice"
+        )
+        assert da.results[0]["enabled"] is True
+
+    def test_parsing_installed_and_enabled(self):
+        da = DumpsysAccessibilityArtifact()
+        file = get_artifact("android_data/dumpsys_accessibility_enabled.txt")
+        with open(file) as f:
+            data = f.read()
+
+        assert len(da.results) == 0
+        da.parse(data)
+        assert len(da.results) == 5
+
+        enabled = [r for r in da.results if r["enabled"]]
+        assert len(enabled) == 1
+        assert enabled[0]["package_name"] == "com.samsung.accessibility"
+        assert (
+            enabled[0]["service"]
+            == "com.samsung.accessibility/.universalswitch.UniversalSwitchService  (A11yTool)"
+        )
+
+        not_enabled = [r for r in da.results if not r["enabled"]]
+        assert len(not_enabled) == 4
 
     def test_ioc_check(self, indicator_file):
         da = DumpsysAccessibilityArtifact()

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:{}