Fix AndroidQF artifact discovery in root-level ZIP archives

Bound the ADB manager state by matching braces

.github/workflows/release.yml

@@ -34,6 +34,7 @@ jobs:
   publish:
     if: github.ref_type == 'tag'
     runs-on: ubuntu-latest
+    environment: pypi # only tag refs may deploy; the PyPI trusted publisher requires it
     permissions:
       id-token: write # PyPI trusted publishing
     steps:

MANIFEST.in

@@ -0,0 +1,4 @@
+prune tests
+prune docs
+prune .github
+exclude uv.lock Dockerfile* mkdocs.yml .readthedocs.yaml .safety-policy.yml .python-version .gitignore Makefile

pyproject.toml

@@ -24,10 +24,8 @@ dependencies = [
     "simplejson==4.1.1",
     "packaging==26.3",
     "appdirs==1.4.4",
-    "iphone_backup_decrypt==0.9.0",
+    "iphone_backup_decrypt==0.10.0",
     "pycryptodome>=3.20.0",
-    "adb-shell[usb]==0.4.4",
-    "libusb1==3.4.0",
     "cryptography==50.0.0",
     "PyYAML>=6.0.2",
     "pyahocorasick==2.3.1",

src/mvt/android/artifacts/dumpsys_accessibility.py

@@ -10,8 +10,45 @@ from .artifact import AndroidArtifact
 
 
 class DumpsysAccessibilityArtifact(AndroidArtifact):
+    # One list for both record shapes — a service record and a count-only
+    # record must stay the same shape.
+    _FIELDS = (
+        "user_id",
+        "component",
+        "package_name",
+        "service_name",
+        "installed",
+        "enabled",
+        "binding",
+        "bound",
+        "crashed",
+        "accessibility_tool",
+        "installed_service_count",
+        "unnamed_service_count",
+    )
+
     def check_indicators(self) -> None:
         for result in self.results:
+            # A stated count the dump does not back with component names is a
+            # coverage statement, not a service: low, but not silent.
+            if not result.get("component"):
+                stated = result["installed_service_count"]
+                unnamed = result["unnamed_service_count"]
+                if unnamed == stated:
+                    detail = "does not list their component names"
+                else:
+                    detail = (
+                        f"lists the component names of only {stated - unnamed} "
+                        f"of them ({unnamed} unnamed)"
+                    )
+                self.alertstore.low(
+                    f"The accessibility dump states {stated} installed "
+                    f"service(s) for user {result['user_id']} but {detail}",
+                    "",
+                    result,
+                )
+                continue
+
             if self.indicators:
                 ioc_match = self.indicators.check_app_id(result["package_name"])
                 if ioc_match:
@@ -20,11 +57,22 @@ class DumpsysAccessibilityArtifact(AndroidArtifact):
                     )
                     continue
 
-            self.alertstore.medium(
-                f'Found accessibility service: "{result["component"]}"',
-                "",
-                result,
+            # Installed is not enabled. A service can sit installed for years
+            # without ever being switched on, and that is the difference this
+            # channel is read for — an alert that says only "found" makes every
+            # device look equally exposed. A service the dump says is switched
+            # OFF is reported LOW, so it still reaches the analyst without
+            # competing with one that is actually running; a dump that does not
+            # state the enabled state stays MEDIUM, because "not stated" is not
+            # "not enabled".
+            message = (
+                f'Found accessibility service: "{result["component"]}" '
+                f"({self._describe_state(result)})"
             )
+            if result.get("enabled") is False and not result.get("bound"):
+                self.alertstore.low(message, "", result)
+            else:
+                self.alertstore.medium(message, "", result)
 
     def parse(self, content: str) -> None:
         """
@@ -36,6 +84,13 @@ class DumpsysAccessibilityArtifact(AndroidArtifact):
 
         self.results: list[dict[str, Any]] = []
         services: dict[tuple[int | None, str], dict] = {}
+        # Which state sections the dump printed, per user: one user's printed
+        # `enabled services` says nothing about another user's.
+        seen_states: dict[int | None, set[str]] = {}
+        # `installedServiceCount=N` from the user's `attributes:{…}` line is on
+        # most builds the only statement about installed services in the dump:
+        # few print the `installed services: {…}` block.
+        installed_counts: dict[int | None, int] = {}
         user_id: int | None = None
         state: str | None = None
 
@@ -44,6 +99,10 @@ class DumpsysAccessibilityArtifact(AndroidArtifact):
             if user_match:
                 user_id = int(user_match.group(1))
 
+            count_match = re.search(r"installedServiceCount=(\d+)", line)
+            if count_match:
+                installed_counts[user_id] = int(count_match.group(1))
+
             stripped = line.strip()
             state_match = re.match(
                 r"(?i)(installed|enabled|binding|bound|crashed) services\s*:\s*\{(.*)",
@@ -51,6 +110,7 @@ class DumpsysAccessibilityArtifact(AndroidArtifact):
             )
             if state_match:
                 state = state_match.group(1).lower()
+                seen_states.setdefault(user_id, set()).add(self._state_field(state))
                 inline = state_match.group(2)
                 for component in re.findall(
                     r"\{?([\w.$-]+/[\w.$-]+)(?:\s+\(A11yTool\))?\}?", inline
@@ -79,24 +139,65 @@ class DumpsysAccessibilityArtifact(AndroidArtifact):
                 service[self._state_field(state)] = True
                 service["accessibility_tool"] = "(A11yTool)" in stripped
 
+        # A section that was never printed is NOT the same as one printed
+        # empty: the first says nothing, the second says nothing is enabled.
+        # Defaulting every flag to False would turn "not stated" into "not
+        # enabled". Flags for sections this dump never printed stay None.
+        named: dict[int | None, int] = {}
+        for (service_user, _component), service in services.items():
+            printed = seen_states.get(service_user, set())
+            for state in ("installed", "enabled", "binding", "bound", "crashed"):
+                if self._state_field(state) not in printed:
+                    service[self._state_field(state)] = None
+            service["installed_service_count"] = installed_counts.get(service_user)
+            named[service_user] = named.get(service_user, 0) + 1
+
         self.results.extend(services.values())
 
+        # A stated count the named services do not add up to would leave no
+        # trace of the rest: the module would log "a total of 0" about a dump
+        # that said five, or "a total of 1" about one that said two.
+        for count_user, count in installed_counts.items():
+            unnamed = count - named.get(count_user, 0)
+            if unnamed > 0:
+                self.results.append(self._new_unlisted(count_user, count, unnamed))
+
+    @staticmethod
+    def _describe_state(result: dict) -> str:
+        if result.get("bound"):
+            return "enabled and bound"
+        if result.get("enabled"):
+            return "enabled"
+        if result.get("enabled") is None:
+            return "installed, enabled state not stated"
+        return "installed, not enabled"
+
     @staticmethod
     def _state_field(state: str) -> str:
         return {"binding": "binding", "bound": "bound"}.get(state, state)
 
+    @staticmethod
+    def _new_unlisted(user_id: int | None, count: int, unnamed: int) -> dict:
+        """The dump's own count for a user, and how many of those services it
+        did not name.
+
+        Every other field stays unknown: the dump named no service to carry it.
+        """
+        record: dict[str, Any] = dict.fromkeys(DumpsysAccessibilityArtifact._FIELDS)
+        record["user_id"] = user_id
+        record["installed_service_count"] = count
+        record["unnamed_service_count"] = unnamed
+        return record
+
     @staticmethod
     def _new_service(component: str, user_id: int | None) -> dict:
-        package_name, service_name = component.split("/", 1)
-        return {
-            "user_id": user_id,
-            "component": component,
-            "package_name": package_name,
-            "service_name": service_name,
-            "installed": False,
-            "enabled": False,
-            "binding": False,
-            "bound": False,
-            "crashed": False,
-            "accessibility_tool": False,
-        }
+        record: dict[str, Any] = dict.fromkeys(
+            DumpsysAccessibilityArtifact._FIELDS, False
+        )
+        record["user_id"] = user_id
+        record["component"] = component
+        record["package_name"], record["service_name"] = component.split("/", 1)
+        # Filled in after parsing: the count is per user.
+        record["installed_service_count"] = None
+        record["unnamed_service_count"] = None
+        return record

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/artifacts/mounts.py

@@ -136,6 +136,10 @@ class Mounts(AndroidArtifact):
             mount_options = fields[5].split(",")
             super_options = fields[separator + 3].split(",")
             options = list(dict.fromkeys(mount_options + super_options))
+            # fields[5] are the per-mount (VFS) flags, the field after the
+            # separator the superblock's; a mount is writable only if both
+            # allow it. The merged list made every read-only overlay a HIGH.
+            is_read_write = "rw" in mount_options and "rw" in super_options
             mount_point = fields[4].replace("\\040", " ")
             device = fields[separator + 2].replace("\\040", " ")
             filesystem_type = fields[separator + 1]
@@ -156,7 +160,7 @@ class Mounts(AndroidArtifact):
                     "options_list": options,
                     "optional_fields": fields[6:separator],
                     "is_system_partition": is_system,
-                    "is_read_write": "rw" in options,
+                    "is_read_write": is_read_write,
                     "process_ids": [process_id],
                 }
             )
@@ -191,7 +195,14 @@ class Mounts(AndroidArtifact):
                     )
 
             # Check for other suspicious mount options
-            suspicious_opts = [opt for opt in options if opt in SUSPICIOUS_OPTIONS]
+            # `rw` sitting in the merged option list is not evidence of
+            # writability (see is_read_write in parse_mountinfo); every other
+            # suspicious option is meaningful in either list.
+            suspicious_opts = [
+                opt
+                for opt in options
+                if opt in SUSPICIOUS_OPTIONS and (opt != "rw" or mount["is_read_write"])
+            ]
             if suspicious_opts and mount["is_system_partition"]:
                 if (
                     "noatime" in mount["mount_options"]

src/mvt/android/cmd_check_bugreport.py

@@ -3,14 +3,19 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
+import fnmatch
+import io
 import logging
 import os
 from pathlib import Path
 from typing import List, Optional
-from zipfile import ZipFile
+from zipfile import BadZipFile, ZipFile
 
 from mvt.android.artifacts.getprop import GetProp
-from mvt.android.modules.bugreport.base import BugReportModule
+from mvt.android.modules.bugreport.base import (
+    DUMPSTATE_ENTRY_POINTS,
+    BugReportModule,
+)
 from mvt.common.command import Command
 from mvt.common.indicators import Indicators
 from mvt.common.module import MVTModule
@@ -88,6 +93,48 @@ class CmdAndroidCheckBugreport(Command):
         for file_name in self.__zip.namelist():
             self.__files.append(file_name)
 
+        if not self._has_dumpstate(self.__files):
+            nested = self._nested_bugreport(bugreport_zip)
+            if nested:
+                self.__zip = nested
+                self.__files = list(nested.namelist())
+
+    @staticmethod
+    def _has_dumpstate(file_names: List[str]) -> bool:
+        """Whether these members carry any of the entry points the bug report
+        modules read (see `BugReportModule._get_dumpstate_file`)."""
+        return any(
+            fnmatch.filter(file_names, pattern) for pattern in DUMPSTATE_ENTRY_POINTS
+        )
+
+    def _nested_bugreport(self, outer: ZipFile) -> Optional[ZipFile]:
+        """Descend one level into an OEM wrapper archive.
+
+        MIUI / HyperOS hands out a zip of app logs, ANR traces and tcpdump
+        captures with the real `bugreport-<device>-<timestamp>.zip` nested
+        inside. Without this descent the outer archive has no entry point,
+        every module reports it found no files, and the command still exits 0
+        with an empty result.
+        """
+        candidates = [
+            name for name in outer.namelist() if name.lower().endswith(".zip")
+        ]
+        candidates.sort(
+            key=lambda name: (
+                "bugreport" not in name.lower() and "dumpstate" not in name.lower()
+            )
+        )
+        for name in candidates:
+            try:
+                inner = ZipFile(io.BytesIO(outer.read(name)))
+            except (BadZipFile, OSError):
+                continue
+            if self._has_dumpstate(inner.namelist()):
+                log.info("Found the bug report nested inside the archive: %s", name)
+                return inner
+            inner.close()
+        return None
+
     def init(self) -> None:
         if self.target_path:
             self.log.info("Checking Android bug report at path: %s", self.target_path)

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/android/modules/bugreport/base.py

@@ -6,6 +6,7 @@ import datetime
 import fnmatch
 import logging
 import os
+import re
 from pathlib import Path
 from typing import List, Optional
 from zipfile import ZipFile
@@ -13,6 +14,15 @@ from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 
 from mvt.common.module import ModuleResults, MVTModule
 
+# `------ 0.101s was the duration of 'SOME SECTION' ------`, printed when that
+# section finishes and not necessarily between two sections.
+SECTION_DURATION = re.compile(r"^-{3,}\s*[0-9.]+s was the duration of", re.IGNORECASE)
+
+# The members a bug report archive can be entered through, in the order
+# _get_dumpstate_file() tries them. An archive carrying none of them is not a
+# bug report at this level (see CmdAndroidCheckBugreport._has_dumpstate).
+DUMPSTATE_ENTRY_POINTS = ("main_entry.txt", "dumpState_*.log", "*/dumpsys.txt")
+
 
 class BugReportModule(MVTModule):
     """This class provides a base for all Android Bug Report modules."""
@@ -87,7 +97,9 @@ class BugReportModule(MVTModule):
         return data
 
     def _get_dumpstate_file(self) -> Optional[bytes]:
-        main = self._get_files_by_pattern("main_entry.txt")
+        main_entry, dumpstate_log, dumpsys_txt = DUMPSTATE_ENTRY_POINTS
+
+        main = self._get_files_by_pattern(main_entry)
         if main:
             main_content = self._get_file_content(main[0])
             try:
@@ -95,11 +107,11 @@ class BugReportModule(MVTModule):
             except KeyError:
                 return None
 
-        dumpstate_logs = self._get_files_by_pattern("dumpState_*.log")
+        dumpstate_logs = self._get_files_by_pattern(dumpstate_log)
         if dumpstate_logs:
             return self._get_file_content(dumpstate_logs[0])
 
-        dumpsys_files = self._get_files_by_pattern("*/dumpsys.txt")
+        dumpsys_files = self._get_files_by_pattern(dumpsys_txt)
         if dumpsys_files:
             return self._get_file_content(dumpsys_files[0])
 
@@ -122,6 +134,12 @@ class BugReportModule(MVTModule):
                     in_section = True
                 continue
             if stripped.startswith("------"):
+                # dumpstate prints a section's timing line when that section
+                # finishes, which can land in the middle of the one being
+                # written. Treating it as a boundary truncates the section at
+                # an arbitrary point, silently.
+                if SECTION_DURATION.match(stripped):
+                    continue
                 break
             lines.append(line)
         return "\n".join(lines)

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

@@ -48,9 +48,22 @@ class DumpsysAccessibility(DumpsysAccessibilityArtifact, BugReportModule):
         )
         self.parse(content)
 
+        listed = unnamed = 0
         for result in self.results:
-            self.log.info('Found accessibility service "%s"', result.get("component"))
+            if result.get("component"):
+                listed += 1
+                self.log.info(
+                    'Found accessibility service "%s"', result.get("component")
+                )
+                continue
+            # The operator gets this per user as a LOW alert from
+            # check_indicators(); here it only has to survive into the summary,
+            # so that a stated count never reads as "a total of 0".
+            unnamed += result.get("unnamed_service_count") or 0
 
         self.log.info(
-            "Identified a total of %d accessibility services", len(self.results)
+            "Identified a total of %d accessibility services (%d more stated by "
+            "the dump without a component name)",
+            listed,
+            unnamed,
         )

src/mvt/common/url.py

@@ -349,7 +349,9 @@ class URL:
             return tld_obj
         if tld_obj is None:
             return ""
-        return tld_obj.parsed_url.netloc.lower().lstrip("www.")
+        # removeprefix, not lstrip: lstrip takes a set of characters, so it ate the
+        # leading "w"s and dots of any domain ("web.evil.com" -> "eb.evil.com").
+        return tld_obj.parsed_url.netloc.lower().removeprefix("www.")
 
     def get_top_level(self) -> str:
         """Get only the top-level domain from a URL.

src/mvt/ios/decrypt.py

@@ -52,7 +52,7 @@ class MVTEncryptedBackup(EncryptedBackup):
 
     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.
+    internals). Pinned to iphone_backup_decrypt==0.10.0.
     """
 
     def __init__(self, *, backup_directory, passphrase=None, derived_key=None):

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

@@ -35,11 +35,16 @@ class TestDumpsysAccessibilityArtifact:
 
         assert len(da.results) == 0
         da.parse(data)
-        assert len(da.results) == 1
+        # One named service, plus one count-only record: the dump states
+        # `installedServiceCount=2` and names only one component.
+        assert len(da.results) == 2
         assert da.results[0]["package_name"] == "com.malware.accessibility"
         assert da.results[0]["service_name"] == "com.malware.service.malwareservice"
         assert da.results[0]["enabled"] is True
-        assert da.results[0]["installed"] is False
+        # This fixture never prints an `installed services:` section, so the
+        # dump does not state the installed status. Reporting False would turn
+        # "not stated" into "not installed", so it reads None here.
+        assert da.results[0]["installed"] is None
 
     def test_accessibility_service_alert(self):
         da = DumpsysAccessibilityArtifact()
@@ -50,9 +55,11 @@ class TestDumpsysAccessibilityArtifact:
 
         da.check_indicators()
 
-        assert len(da.alertstore.alerts) == 1
+        assert len(da.alertstore.alerts) == 2
         assert da.alertstore.alerts[0].level == AlertLevel.MEDIUM
         assert da.alertstore.alerts[0].event == da.results[0]
+        assert da.alertstore.alerts[1].level == AlertLevel.LOW
+        assert da.alertstore.alerts[1].event == da.results[1]
 
     def test_same_component_is_kept_for_each_user(self):
         da = DumpsysAccessibilityArtifact()
@@ -84,7 +91,11 @@ User state[attributes:{id=10
         assert len(da.alertstore.alerts) == 0
         da.check_indicators()
         assert len(da.alertstore.alerts) == len(da.results)
-        assert da.alertstore.count(AlertLevel.MEDIUM) == 3
+        # Every service in this fixture is installed and switched off
+        # (`enabled services:{}` is printed and empty), so the three non-IOC
+        # findings are LOW, not MEDIUM. The IOC match is unaffected by the
+        # state.
+        assert da.alertstore.count(AlertLevel.LOW) == 3
         assert da.alertstore.count(AlertLevel.CRITICAL) == 1
         critical_alert = next(
             alert

tests/android/test_artifact_dumpsys_accessibility_count.py

@@ -0,0 +1,149 @@
+# 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/
+"""The dump's own installed-service count must survive into the artifact.
+
+Most builds never print the `installed services: {…}` block; they state `installedServiceCount=N` in the user's `attributes:{…}` line and
+list nothing. Dropping that number makes an artifact that says "no
+accessibility services" about a dump that said there are five.
+"""
+
+from mvt.android.artifacts.dumpsys_accessibility import DumpsysAccessibilityArtifact
+from mvt.common.alerts import AlertLevel
+
+from ..utils import get_artifact
+
+AOSP_NO_LIST = """\
+ACCESSIBILITY MANAGER (dumpsys accessibility)
+User state[
+     attributes:{id=0, touchExplorationEnabled=false, installedServiceCount=5}
+     Bound services:{}
+     Enabled services:{}
+     Binding services:{}
+     Crashed services:{}
+"""
+
+ONE_UI_WITH_LIST = """\
+ACCESSIBILITY MANAGER (dumpsys accessibility)
+User state[attributes:{id=0, installedServiceCount=2}
+  installed services: {
+    0 : com.example.app/com.example.app.Service
+    1 : com.other.app/.Helper
+  }
+  enabled services: {
+  }
+"""
+
+TWO_USERS = """\
+ACCESSIBILITY MANAGER (dumpsys accessibility)
+User state[attributes:{id=0, installedServiceCount=1}
+  installed services: {
+    0 : com.example.app/com.example.app.Service
+  }
+User state[attributes:{id=95, installedServiceCount=3}
+     Enabled services:{}
+"""
+
+PARTIAL_TWO_USERS = """\
+ACCESSIBILITY MANAGER (dumpsys accessibility)
+User state[attributes:{id=0, installedServiceCount=3}
+     Enabled services:{{com.example.app/com.example.app.Service}}
+User state[attributes:{id=10, installedServiceCount=1}
+     Enabled services:{{com.other.app/.Helper}}
+"""
+
+ZERO_COUNT = """\
+ACCESSIBILITY MANAGER (dumpsys accessibility)
+User state[attributes:{id=0, installedServiceCount=0}
+     Enabled services:{}
+"""
+
+
+def _parse(content):
+    artifact = DumpsysAccessibilityArtifact()
+    artifact.results = []
+    artifact.parse(content)
+    return artifact
+
+
+class TestAccessibilityInstalledServiceCount:
+    def test_stated_count_without_a_list_is_kept(self):
+        artifact = _parse(AOSP_NO_LIST)
+        assert len(artifact.results) == 1
+        record = artifact.results[0]
+        assert record["installed_service_count"] == 5
+        assert record["component"] is None
+        # Every state flag stays unknown: the dump named no service to which a
+        # state could belong.
+        assert record["installed"] is None
+        assert record["enabled"] is None
+
+    def test_a_stated_count_is_reported_as_low(self):
+        artifact = _parse(AOSP_NO_LIST)
+        artifact.check_indicators()
+        alerts = artifact.alertstore.alerts
+        assert len(alerts) == 1
+        # A count is a coverage statement, not a running service: it must not
+        # compete with a service the dump says is enabled.
+        assert alerts[0].level == AlertLevel.LOW
+        assert "does not list their component names" in alerts[0].message
+        assert "5 installed" in alerts[0].message
+
+    def test_a_listed_user_carries_the_count_on_each_service(self):
+        artifact = _parse(ONE_UI_WITH_LIST)
+        assert len(artifact.results) == 2
+        assert {record["installed_service_count"] for record in artifact.results} == {2}
+        assert all(record["component"] for record in artifact.results)
+
+    def test_only_the_unlisted_user_gets_a_count_record(self):
+        artifact = _parse(TWO_USERS)
+        listed = [record for record in artifact.results if record["component"]]
+        unlisted = [record for record in artifact.results if not record["component"]]
+        assert [record["user_id"] for record in listed] == [0]
+        assert [record["user_id"] for record in unlisted] == [95]
+        assert unlisted[0]["installed_service_count"] == 3
+
+    def test_a_zero_count_adds_nothing(self):
+        # "Zero installed" is a negative result the empty section already
+        # states; a record for it would be noise.
+        assert _parse(ZERO_COUNT).results == []
+
+    def test_a_partly_named_count_reports_the_unnamed_rest(self):
+        # The Android 14 fixture states `installedServiceCount=2` and names one
+        # enabled component. The listing is incomplete, and must not read as
+        # complete.
+        artifact = DumpsysAccessibilityArtifact()
+        artifact.results = []
+        with open(
+            get_artifact("android_data/dumpsys_accessibility_v14_or_later.txt")
+        ) as handle:
+            artifact.parse(handle.read())
+        unlisted = [record for record in artifact.results if not record["component"]]
+        assert len(unlisted) == 1
+        assert unlisted[0]["installed_service_count"] == 2
+        assert unlisted[0]["unnamed_service_count"] == 1
+
+        artifact.check_indicators()
+        low = [
+            alert
+            for alert in artifact.alertstore.alerts
+            if alert.level == AlertLevel.LOW
+        ]
+        assert len(low) == 1
+        assert "only 1 of them (1 unnamed)" in low[0].message
+
+    def test_the_unnamed_rest_is_counted_per_user(self):
+        # User 0 names one of three, user 10 names its only one: the gap
+        # belongs to user 0 alone.
+        artifact = _parse(PARTIAL_TWO_USERS)
+        unlisted = [record for record in artifact.results if not record["component"]]
+        assert [
+            (record["user_id"], record["unnamed_service_count"]) for record in unlisted
+        ] == [(0, 2)]
+
+    def test_a_fully_named_count_adds_nothing(self):
+        artifact = _parse(ONE_UI_WITH_LIST)
+        assert all(
+            record["unnamed_service_count"] is None for record in artifact.results
+        )

tests/android/test_artifact_dumpsys_accessibility_state.py

@@ -0,0 +1,155 @@
+# 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/
+"""Installed is not the same as enabled, and the dump says which.
+
+Two things this guards:
+
+  * a section the dump never printed must read as None ("not stated"), not as
+    False ("not enabled");
+  * the alert must name the state, instead of firing identically on a device
+    where nothing is switched on and one where something is bound.
+"""
+
+from types import SimpleNamespace
+
+from mvt.android.artifacts.dumpsys_accessibility import DumpsysAccessibilityArtifact
+from mvt.common.alerts import AlertLevel
+
+from ..utils import get_artifact
+
+NO_STATE_SECTIONS = """\
+ACCESSIBILITY MANAGER (dumpsys accessibility)
+User state[attributes:{id=0, currentUser=true}
+  installed services: {
+    0 : com.example.app/com.example.app.Service
+  }
+"""
+
+ENABLED_BLOCK = """\
+ACCESSIBILITY MANAGER (dumpsys accessibility)
+User state[attributes:{id=0, currentUser=true}
+  installed services: {
+    0 : com.example.app/com.example.app.Service
+    1 : com.other.app/.Helper
+  }
+  enabled services: {
+    0 : com.other.app/.Helper
+  }
+  bound services:{
+    0 : com.other.app/.Helper
+  }
+"""
+
+# User 0 prints only `installed services`, user 10 only `Enabled services`.
+# Neither section speaks for the other user.
+TWO_USERS_DIFFERENT_SECTIONS = """\
+ACCESSIBILITY MANAGER (dumpsys accessibility)
+User state[attributes:{id=0, currentUser=true}
+  installed services: {
+    0 : com.example.app/com.example.app.Service
+  }
+User state[attributes:{id=10, currentUser=false}
+     Enabled services:{{com.other.app/.Helper}}
+"""
+
+
+class _IndicatorsMatching:
+    """Minimal stand-in: matches one package id, like the STIX2 loader would."""
+
+    def __init__(self, package_name: str) -> None:
+        self.package_name = package_name
+
+    def check_app_id(self, app_id):
+        if app_id != self.package_name:
+            return None
+        return SimpleNamespace(
+            message=f"Found a known suspicious app: {app_id}", ioc={"value": app_id}
+        )
+
+
+class TestAccessibilityServiceState:
+    def _parse(self, content):
+        artifact = DumpsysAccessibilityArtifact()
+        artifact.results = []
+        artifact.parse(content)
+        return {r["component"]: r for r in artifact.results}
+
+    def test_absent_sections_leave_the_state_unknown(self):
+        # None, not False: a build that does not print the sections says
+        # nothing about what is enabled, and that must not read as "nothing".
+        state = self._parse(NO_STATE_SECTIONS)[
+            "com.example.app/com.example.app.Service"
+        ]
+        assert state["installed"] is True
+        assert state["enabled"] is None
+        assert state["bound"] is None
+
+    def test_enabled_and_bound_are_attributed_per_service(self):
+        results = self._parse(ENABLED_BLOCK)
+        installed_only = results["com.example.app/com.example.app.Service"]
+        active = results["com.other.app/.Helper"]
+        assert (installed_only["enabled"], installed_only["bound"]) == (False, False)
+        assert (active["enabled"], active["bound"]) == (True, True)
+
+    def test_alert_message_carries_the_state(self):
+        artifact = DumpsysAccessibilityArtifact()
+        artifact.results = []
+        with open(get_artifact("android_data/dumpsys_accessibility.txt")) as handle:
+            artifact.parse(handle.read())
+        artifact.check_indicators()
+        assert artifact.alertstore.alerts
+        assert all("installed" in alert.message for alert in artifact.alertstore.alerts)
+
+    def test_a_switched_off_service_is_low_and_a_running_one_medium(self):
+        # A service the dump says is OFF still reaches the analyst, but must not
+        # compete with one that is actually bound. "Not stated" is not "off".
+        artifact = DumpsysAccessibilityArtifact()
+        artifact.results = []
+        artifact.parse(ENABLED_BLOCK)
+        artifact.check_indicators()
+        by_level = {}
+        for alert in artifact.alertstore.alerts:
+            by_level.setdefault(alert.level, []).append(alert.message)
+        assert artifact.alertstore.count(AlertLevel.LOW) == 1
+        assert artifact.alertstore.count(AlertLevel.MEDIUM) == 1
+        assert "com.example.app" in by_level[AlertLevel.LOW][0]
+        assert "com.other.app" in by_level[AlertLevel.MEDIUM][0]
+
+    def test_an_unstated_enabled_state_stays_medium(self):
+        artifact = DumpsysAccessibilityArtifact()
+        artifact.results = []
+        artifact.parse(NO_STATE_SECTIONS)
+        artifact.check_indicators()
+        assert artifact.alertstore.count(AlertLevel.MEDIUM) == 1
+        assert artifact.alertstore.count(AlertLevel.LOW) == 0
+
+    def test_a_disabled_service_is_still_matched_against_indicators(self):
+        # The state decides the severity of an ordinary finding, never whether
+        # the package is compared with the IOC feeds.
+        artifact = DumpsysAccessibilityArtifact()
+        artifact.results = []
+        artifact.parse(ENABLED_BLOCK)
+        artifact.indicators = _IndicatorsMatching("com.example.app")
+        artifact.check_indicators()
+        assert artifact.alertstore.count(AlertLevel.CRITICAL) == 1
+        assert artifact.alertstore.count(AlertLevel.LOW) == 0
+
+    def test_printed_sections_are_tracked_per_user(self):
+        artifact = DumpsysAccessibilityArtifact()
+        artifact.results = []
+        artifact.parse(TWO_USERS_DIFFERENT_SECTIONS)
+        by_user = {r["user_id"]: r for r in artifact.results}
+        # User 0's enabled state is not stated, so it is unknown, not off.
+        assert (by_user[0]["installed"], by_user[0]["enabled"]) == (True, None)
+        # User 10's installed state is not stated either.
+        assert (by_user[10]["installed"], by_user[10]["enabled"]) == (None, True)
+
+        artifact.check_indicators()
+        assert artifact.alertstore.count(AlertLevel.LOW) == 0
+        assert artifact.alertstore.count(AlertLevel.MEDIUM) == 2
+        assert not any(
+            "installed, not enabled" in alert.message
+            for alert in artifact.alertstore.alerts
+        )

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

@@ -0,0 +1,43 @@
+# 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/
+"""A mount is writable only if both mountinfo option layers allow it."""
+
+from mvt.android.artifacts.mounts import Mounts
+from mvt.common.alerts import AlertLevel
+
+# fields[5] is the per-mount (VFS) layer, the field after "-" the superblock.
+RO_OVERLAY = (
+    "304 166 0:130 / /product/usr rw,relatime shared:63 - overlay overlay "
+    "ro,seclabel,lowerdir=/mnt/vendor/ext/product/usr:/product/usr"
+)
+RW_SYSTEM = (
+    "305 166 0:131 / /system rw,relatime shared:64 - ext4 /dev/block/dm-1 "
+    "rw,seclabel,errors=panic"
+)
+
+
+def _run(line):
+    mounts = Mounts()
+    mounts.results = Mounts.parse_mountinfo(line, 1)
+    mounts.check_indicators()
+    return mounts
+
+
+class TestMountsReadWriteLayers:
+    def test_rw_vfs_over_ro_superblock_is_not_writable(self):
+        mounts = _run(RO_OVERLAY)
+        assert mounts.results[0]["is_read_write"] is False
+        assert mounts.alertstore.count(AlertLevel.HIGH) == 0
+        assert mounts.alertstore.count(AlertLevel.MEDIUM) == 0
+
+    def test_rw_in_both_layers_still_alerts(self):
+        mounts = _run(RW_SYSTEM)
+        assert mounts.results[0]["is_read_write"] is True
+        assert mounts.alertstore.count(AlertLevel.HIGH) == 1
+
+    def test_both_option_layers_are_still_reported(self):
+        mounts = _run(RO_OVERLAY)
+        options = mounts.results[0]["options_list"]
+        assert "rw" in options and "ro" in options

tests/android/test_bugreport_section_extraction.py

@@ -0,0 +1,42 @@
+# 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/
+"""A section ends at the next section, not at a timing line printed inside it."""
+
+from mvt.android.modules.bugreport.base import BugReportModule
+
+# dumpstate prints a section's duration when that section finishes, which can
+# land in the middle of the section currently being written.
+DUMPSTATE = """\
+------ SYSTEM PROPERTIES (getprop) ------
+[nfc.initialized]: [true]
+------ 0.101s was the duration of 'DROPBOX SYSTEM SERVER CRASHES' ------
+[ro.build.version.sdk]: [30]
+[ro.product.model]: [SM-A305F]
+------ 0.064s was the duration of 'SYSTEM PROPERTIES' ------
+------ STORAGE INFO (df) ------
+/dev/root  2.9G
+"""
+
+
+class TestExtractCommandSection:
+    def test_a_foreign_timing_line_does_not_end_the_section(self):
+        section = BugReportModule.extract_command_section(
+            DUMPSTATE, "------ SYSTEM PROPERTIES"
+        )
+        assert "[ro.product.model]: [SM-A305F]" in section
+        assert section.count("\n") == 2
+
+    def test_the_next_section_is_still_the_boundary(self):
+        section = BugReportModule.extract_command_section(
+            DUMPSTATE, "------ SYSTEM PROPERTIES"
+        )
+        assert "STORAGE INFO" not in section
+        assert "/dev/root" not in section
+
+    def test_timing_lines_are_not_returned_as_content(self):
+        section = BugReportModule.extract_command_section(
+            DUMPSTATE, "------ SYSTEM PROPERTIES"
+        )
+        assert "was the duration of" not in section

tests/android/test_check_bugreport_wrapper.py

@@ -0,0 +1,56 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2023 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/
+"""An OEM wrapper archive must not read as an empty bug report.
+
+MIUI / HyperOS hands out a zip of app logs with the real
+`bugreport-<device>-<timestamp>.zip` nested inside. The outer archive has
+none of the entry points the modules read, so every module
+reported it found nothing and the command still exited 0 — an empty analysis
+that looks like a finished one.
+"""
+
+import io
+import zipfile
+
+from mvt.android.cmd_check_bugreport import CmdAndroidCheckBugreport
+
+DUMPSTATE = "== dumpstate: 2026-01-01 00:00:00\nDUMP OF SERVICE package:\n"
+
+
+def _inner_zip() -> bytes:
+    buffer = io.BytesIO()
+    with zipfile.ZipFile(buffer, "w") as inner:
+        inner.writestr("main_entry.txt", "bugreport-test-2026-01-01-00-00-00.txt")
+        inner.writestr("bugreport-test-2026-01-01-00-00-00.txt", DUMPSTATE)
+    return buffer.getvalue()
+
+
+def _wrapper_zip() -> zipfile.ZipFile:
+    buffer = io.BytesIO()
+    with zipfile.ZipFile(buffer, "w") as outer:
+        outer.writestr("app_logs/hilog.txt", "unrelated OEM log\n")
+        outer.writestr("bugreport-test-2026-01-01-00-00-00.zip", _inner_zip())
+    return zipfile.ZipFile(io.BytesIO(buffer.getvalue()))
+
+
+class TestCheckBugreportWrapper:
+    def test_nested_bugreport_is_used(self, tmp_path):
+        cmd = CmdAndroidCheckBugreport(results_path=str(tmp_path))
+        cmd.from_zip(_wrapper_zip())
+        module = cmd.modules[0](results_path=str(tmp_path))
+        cmd.module_init(module)
+        assert "main_entry.txt" in module.zip_files
+
+    def test_plain_bugreport_is_left_alone(self, tmp_path):
+        buffer = io.BytesIO()
+        with zipfile.ZipFile(buffer, "w") as archive:
+            archive.writestr("main_entry.txt", "bugreport.txt")
+            archive.writestr("bugreport.txt", DUMPSTATE)
+            archive.writestr("attachments/extra.zip", _inner_zip())
+        cmd = CmdAndroidCheckBugreport(results_path=str(tmp_path))
+        cmd.from_zip(zipfile.ZipFile(io.BytesIO(buffer.getvalue())))
+        module = cmd.modules[0](results_path=str(tmp_path))
+        cmd.module_init(module)
+        assert "attachments/extra.zip" in module.zip_files

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/common/test_indicators.py

@@ -311,3 +311,31 @@ class TestIndicators:
         ind = Indicators(log=logging)
         ind.load_indicators_files([], load_default=False)
         assert ind.total_ioc_count == 9
+
+    def test_check_url_matches_w_prefixed_subdomain(self, tmp_path):
+        import json
+
+        stix_file = tmp_path / "w-domain.stix2"
+        stix_file.write_text(
+            json.dumps(
+                {
+                    "objects": [
+                        {
+                            "type": "indicator",
+                            "pattern": "[domain-name:value = 'web.evil.com']",
+                        }
+                    ]
+                }
+            ),
+            encoding="utf-8",
+        )
+        ind = Indicators(log=logging)
+        ind.load_indicators_files([str(stix_file)], load_default=False)
+
+        for url in (
+            "https://web.evil.com/path",
+            "https://www.web.evil.com/path",
+        ):
+            match = ind.check_url(url)
+            assert match is not None
+            assert match.ioc.value == "web.evil.com"

tests/common/test_url.py

@@ -22,3 +22,23 @@ def test_google_maps_url_is_not_shortened(url):
 
 def test_other_google_short_url_is_shortened():
     assert URL("https://goo.gl/example").check_if_shortened() is True
+
+
+@pytest.mark.parametrize(
+    "url, domain",
+    [
+        ("https://www.example.com/path", "example.com"),
+        # Only the whole "www." prefix comes off, not any leading "w" or "." character.
+        ("https://web.example.com", "web.example.com"),
+        ("https://wow.com", "wow.com"),
+        ("https://wired.com", "wired.com"),
+        ("https://www.wow.com", "wow.com"),
+    ],
+)
+def test_get_domain_strips_only_a_whole_www_prefix(url, domain):
+    assert URL(url).domain == domain
+
+
+def test_shortener_starting_with_w_is_detected():
+    assert URL("https://w3t.org/example").check_if_shortened() is True
+    assert URL("https://www.w3t.org/example").check_if_shortened() is True

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

uv.lock

@@ -6,22 +6,6 @@ resolution-markers = [
     "python_full_version < '3.15'",
 ]
 
-[[package]]
-name = "adb-shell"
-version = "0.4.4"
-source = { registry = "https://pypi.org/simple" }
-dependencies = [
-    { name = "cryptography" },
-    { name = "pyasn1" },
-    { name = "rsa" },
-]
-sdist = { url = "https://files.pythonhosted.org/packages/8f/73/d246034db6f3e374dad9a35ee3f61345a6b239d4febd2a41ab69df9936fe/adb_shell-0.4.4.tar.gz", hash = "sha256:04c305f30a2ca25d5c54b3cd6ce9bb64c36e5f07967b23b3fb6aaecc851b90b6", size = 61822, upload-time = "2023-09-01T03:48:40.348Z" }
-
-[package.optional-dependencies]
-usb = [
-    { name = "libusb1" },
-]
-
 [[package]]
 name = "annotated-types"
 version = "0.8.0"
@@ -625,14 +609,14 @@ wheels = [
 
 [[package]]
 name = "iphone-backup-decrypt"
-version = "0.9.0"
+version = "0.10.0"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "pycryptodome" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/6f/e7/bcdacdec21d628122ba240e7f742ab2175149e58672be63af55ff37a0f28/iphone_backup_decrypt-0.9.0.tar.gz", hash = "sha256:13b18fef3c8e3af627914f8c1a429bbc5555dfb0505239ba49efe99984cc0c96", size = 16125, upload-time = "2024-09-18T15:50:12.179Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/01/b8/a2ec2b10d792403d36bb6661fea6357179c31ece1ed3285c442e636d6335/iphone_backup_decrypt-0.10.0.tar.gz", hash = "sha256:5f2799e58f908cce28d7929dd82231abc8acd57c25b37bce6a2d6a82e018d343", size = 20000, upload-time = "2026-09-17T21:46:44Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/b8/94/64a31be93f72e0a254bde68e4cf7d24aef37a0a985754a197fa1b028a665/iphone_backup_decrypt-0.9.0-py3-none-any.whl", hash = "sha256:55b5adfafac757f58aa6444b83a4cc2c20cdd699c6ff1d2f4b549936a5dad92c", size = 15767, upload-time = "2024-09-18T15:50:10.537Z" },
+    { url = "https://files.pythonhosted.org/packages/89/10/349b2a720e012921d1457e1d1c2b921b16300a5f3570c4c64d0d3787539f/iphone_backup_decrypt-0.10.0-py3-none-any.whl", hash = "sha256:1ac4345a07b104a3eacab20932b8621cd1e0164e0f03d913e66eed6ea5ee173e", size = 17400, upload-time = "2026-09-17T21:46:42Z" },
 ]
 
 [[package]]
@@ -777,17 +761,6 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/38/a6/800800bfed7b1fb10fc3f3d557785c3854e80d3f7a9800d784b176a1fc2d/librt-0.15.0-cp315-cp315t-win_arm64.whl", hash = "sha256:84d244b00604d17df3fc7736c327892d6bba66181254aa4087be807b6c342bdc", size = 110700, upload-time = "2026-08-07T10:49:15.499Z" },
 ]
 
-[[package]]
-name = "libusb1"
-version = "3.4.0"
-source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/89/35/f9d2a990d092d647b47540cd229e1d68432c0f51183484ca189612a4824c/libusb1-3.4.0.tar.gz", hash = "sha256:9cf5638506d54f21bf36550d97ea63189111a23c4d8078f630103a2052135f45", size = 91206, upload-time = "2026-05-16T20:59:19.315Z" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/d4/64/d4b59444e4d3b6979aa5eb58840634465a24b41a9ab03dcf8434c9b89551/libusb1-3.4.0-py3-none-any.whl", hash = "sha256:e83d034e44c3efe1c4599c6281d34bca50a38c12cab3b7b6217d583161a01ffd", size = 67373, upload-time = "2026-05-16T20:59:13.142Z" },
-    { url = "https://files.pythonhosted.org/packages/7e/55/a838a4278fac4ee49bf670ddc59d621999a1cb2a3188fb13a23609ea4f06/libusb1-3.4.0-py3-none-win32.whl", hash = "sha256:0a1aa1416034690eb9dc9a895eda0fef44d853bca1053eb1de50a5906684846d", size = 129704, upload-time = "2026-05-16T20:59:15.243Z" },
-    { url = "https://files.pythonhosted.org/packages/5f/08/02aecf6dad627534a5835244ece14d7f187ef430354d9b8551199934d059/libusb1-3.4.0-py3-none-win_amd64.whl", hash = "sha256:b7dcc1f324a895af6aac708bc5513a17373f97349cc2ab8a277519c788bd18ca", size = 141212, upload-time = "2026-05-16T20:59:17.286Z" },
-]
-
 [[package]]
 name = "markdown"
 version = "3.10.3"
@@ -1016,13 +989,11 @@ wheels = [
 name = "mvt"
 source = { editable = "." }
 dependencies = [
-    { name = "adb-shell", extra = ["usb"] },
     { name = "appdirs" },
     { name = "betterproto2" },
     { name = "click" },
     { name = "cryptography" },
     { name = "iphone-backup-decrypt" },
-    { name = "libusb1" },
     { name = "nskeyedunarchiver" },
     { name = "packaging" },
     { name = "pyahocorasick" },
@@ -1063,13 +1034,11 @@ docs = [
 
 [package.metadata]
 requires-dist = [
-    { name = "adb-shell", extras = ["usb"], specifier = "==0.4.4" },
     { name = "appdirs", specifier = "==1.4.4" },
     { name = "betterproto2", specifier = "==0.10.0" },
     { name = "click", specifier = "==8.4.2" },
     { name = "cryptography", specifier = "==50.0.0" },
-    { name = "iphone-backup-decrypt", specifier = "==0.9.0" },
-    { name = "libusb1", specifier = "==3.4.0" },
+    { name = "iphone-backup-decrypt", specifier = "==0.10.0" },
     { name = "nskeyedunarchiver", specifier = "==1.5.2" },
     { name = "packaging", specifier = "==26.3" },
     { name = "pyahocorasick", specifier = "==2.3.1" },
@@ -1274,15 +1243,6 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/c5/96/37c50ac951bb0260ec38d8d12e5b51587ef1ef4035c279088f2771544b28/pyahocorasick-2.3.1-cp314-cp314-win_amd64.whl", hash = "sha256:4acb11a0a2ff10519465749d22ad70789e9fe7f81dc8fe9957a8868e499e18ab", size = 35987, upload-time = "2026-04-27T16:32:07.08Z" },
 ]
 
-[[package]]
-name = "pyasn1"
-version = "0.6.4"
-source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/a4/9a/23310166d960def5897e91fe20e5b724601b02a22e84ba1f94232c0b7f67/pyasn1-0.6.4.tar.gz", hash = "sha256:9c447d8431c947fe4c8febc4ed9e760bc29011a5b01e5c74b67025bd9fb8ce81", size = 151262, upload-time = "2026-07-09T01:12:33.988Z" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/9a/3b/6163796d69c3977d1e4287bea4a6979161cbbdd170ebb430511e8e1999ce/pyasn1-0.6.4-py3-none-any.whl", hash = "sha256:deda9277cfd454080ec40b207fb6df82206a3a2688735233cdcd8d3d565f088b", size = 84410, upload-time = "2026-07-09T01:12:32.92Z" },
-]
-
 [[package]]
 name = "pycparser"
 version = "3.0"
@@ -1684,18 +1644,6 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/82/3b/64d4899d73f91ba49a8c18a8ff3f0ea8f1c1d75481760df8c68ef5235bf5/rich-15.0.0-py3-none-any.whl", hash = "sha256:33bd4ef74232fb73fe9279a257718407f169c09b78a87ad3d296f548e27de0bb", size = 310654, upload-time = "2026-04-12T08:24:02.83Z" },
 ]
 
-[[package]]
-name = "rsa"
-version = "4.9.1"
-source = { registry = "https://pypi.org/simple" }
-dependencies = [
-    { name = "pyasn1" },
-]
-sdist = { url = "https://files.pythonhosted.org/packages/da/8a/22b7beea3ee0d44b1916c0c1cb0ee3af23b700b6da9f04991899d0c555d4/rsa-4.9.1.tar.gz", hash = "sha256:e7bdbfdb5497da4c07dfd35530e1a902659db6ff241e39d9953cad06ebd0ae75", size = 29034, upload-time = "2025-04-16T09:51:18.218Z" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/64/8d/0133e4eb4beed9e425d9a98ed6e081a55d195481b7632472be1af08d2f6b/rsa-4.9.1-py3-none-any.whl", hash = "sha256:68635866661c6836b8d39430f97a996acbd61bfa49406748ea243539fe239762", size = 34696, upload-time = "2025-04-16T09:51:17.142Z" },
-]
-
 [[package]]
 name = "ruff"
 version = "0.16.2"