fix: recover complete files from truncated sysdiagnose archives

.github/workflows/release.yml

@@ -34,7 +34,6 @@ 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

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

pyproject.toml

@@ -24,8 +24,10 @@ dependencies = [
     "simplejson==4.1.1",
     "packaging==26.3",
     "appdirs==1.4.4",
-    "iphone_backup_decrypt==0.10.0",
+    "iphone_backup_decrypt==0.9.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,45 +10,8 @@ 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:
@@ -57,22 +20,11 @@ class DumpsysAccessibilityArtifact(AndroidArtifact):
                     )
                     continue
 
-            # 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)})"
+            self.alertstore.medium(
+                f'Found accessibility service: "{result["component"]}"',
+                "",
+                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:
         """
@@ -84,13 +36,6 @@ 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
 
@@ -99,10 +44,6 @@ 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*\{(.*)",
@@ -110,7 +51,6 @@ 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
@@ -139,65 +79,24 @@ 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:
-        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
+        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,
+        }

src/mvt/android/artifacts/dumpsys_adb.py

@@ -29,21 +29,11 @@ class DumpsysADBArtifact(AndroidArtifact):
         stack = [res]
         cur_indent = 0
         in_multiline = False
-        for line in dump_data.strip(b"\r\n").split(b"\n"):
-            line = line.removesuffix(b"\r")
+        for line in dump_data.strip(b"\n").split(b"\n"):
             # 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:
@@ -73,12 +63,6 @@ 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
 
@@ -177,26 +161,6 @@ 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
@@ -216,16 +180,12 @@ class DumpsysADBArtifact(AndroidArtifact):
             self.log.error("Unable to find ADB manager state in dumpsys output")
             return
 
-        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"
-            )
+        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")
             return
 
-        # 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()
+        json_content = content[start_of_json + 2 : end_of_json - 2].rstrip()
 
         parsed = self.indented_dump_parser(json_content)
         if parsed.get("debugging_manager") is None:

src/mvt/android/artifacts/dumpsys_battery_daily.py

@@ -73,12 +73,8 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact):
                 continue
 
             line = line.strip().replace("Update ", "")
-            # 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
+            package_name, vers = line.split(" ", 1)
+            vers_raw = vers.split("=", 1)[1]
             try:
                 version_code: int | str = int(vers_raw)
             except ValueError:

src/mvt/android/artifacts/mounts.py

@@ -136,10 +136,6 @@ 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]
@@ -160,7 +156,7 @@ class Mounts(AndroidArtifact):
                     "options_list": options,
                     "optional_fields": fields[6:separator],
                     "is_system_partition": is_system,
-                    "is_read_write": is_read_write,
+                    "is_read_write": "rw" in options,
                     "process_ids": [process_id],
                 }
             )
@@ -195,14 +191,7 @@ class Mounts(AndroidArtifact):
                     )
 
             # Check for other suspicious mount 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"])
-            ]
+            suspicious_opts = [opt for opt in options if opt in SUSPICIOUS_OPTIONS]
             if suspicious_opts and mount["is_system_partition"]:
                 if (
                     "noatime" in mount["mount_options"]

src/mvt/android/cmd_check_bugreport.py

@@ -3,19 +3,14 @@
 # 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 BadZipFile, ZipFile
+from zipfile import ZipFile
 
 from mvt.android.artifacts.getprop import GetProp
-from mvt.android.modules.bugreport.base import (
-    DUMPSTATE_ENTRY_POINTS,
-    BugReportModule,
-)
+from mvt.android.modules.bugreport.base import BugReportModule
 from mvt.common.command import Command
 from mvt.common.indicators import Indicators
 from mvt.common.module import MVTModule
@@ -93,48 +88,6 @@ 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/bugreport/base.py

@@ -6,7 +6,6 @@ import datetime
 import fnmatch
 import logging
 import os
-import re
 from pathlib import Path
 from typing import List, Optional
 from zipfile import ZipFile
@@ -14,15 +13,6 @@ 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."""
@@ -97,9 +87,7 @@ class BugReportModule(MVTModule):
         return data
 
     def _get_dumpstate_file(self) -> Optional[bytes]:
-        main_entry, dumpstate_log, dumpsys_txt = DUMPSTATE_ENTRY_POINTS
-
-        main = self._get_files_by_pattern(main_entry)
+        main = self._get_files_by_pattern("main_entry.txt")
         if main:
             main_content = self._get_file_content(main[0])
             try:
@@ -107,11 +95,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])
 
@@ -134,12 +122,6 @@ 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,28 +48,9 @@ class DumpsysAccessibility(DumpsysAccessibilityArtifact, BugReportModule):
         )
         self.parse(content)
 
-        listed = unnamed = enabled_count = 0
         for result in self.results:
-            if result.get("component"):
-                listed += 1
-                enabled = result.get("enabled")
-                if enabled is True:
-                    enabled_count += 1
-                self.log.info(
-                    'Found accessibility service "%s" (enabled: %s)',
-                    result.get("component"),
-                    "not stated" if enabled is None else enabled,
-                )
-                continue
-            # The operator gets this per user as a LOW alert from
-            # 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('Found accessibility service "%s"', result.get("component"))
 
         self.log.info(
-            "Identified a total of %d accessibility services, %d reported enabled "
-            "(%d more stated by the dump without a component name)",
-            listed,
-            enabled_count,
-            unnamed,
+            "Identified a total of %d accessibility services", len(self.results)
         )

src/mvt/common/url.py

@@ -349,9 +349,7 @@ class URL:
             return tld_obj
         if tld_obj is None:
             return ""
-        # 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.")
+        return tld_obj.parsed_url.netloc.lower().lstrip("www.")
 
     def get_top_level(self) -> str:
         """Get only the top-level domain from a URL.

src/mvt/ios/cmd_check_sysdiagnose.py

@@ -3,6 +3,7 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
+import gzip
 import json
 import logging
 import os
@@ -11,7 +12,7 @@ import sys
 import tarfile
 import zlib
 from pathlib import Path, PurePosixPath
-from tempfile import TemporaryDirectory
+from tempfile import NamedTemporaryFile, TemporaryDirectory
 from typing import Any, Optional
 
 from mvt.common.command import Command
@@ -23,6 +24,34 @@ from .modules.sysdiagnose import SYSDIAGNOSE_MODULES
 log = logging.getLogger(__name__)
 
 
+class _StreamingGzipFile(gzip.GzipFile):
+    """Deliver available bytes before reporting a missing gzip trailer.
+
+    GzipFile.read(n) tries to fill the entire request. At a truncated end it
+    can raise EOFError even after decoding bytes for complete tar members.
+    read1() returns those bytes first and raises on the following read. The
+    tar stream accepts short reads, so it can retain every complete member.
+    """
+
+    def read(self, size: Optional[int] = -1) -> bytes:
+        return self.read1(-1 if size is None else size)
+
+
+class _SysdiagnoseTarInfo(tarfile.TarInfo):
+    @classmethod
+    def fromtarfile(cls, archive: tarfile.TarFile) -> "_SysdiagnoseTarInfo":
+        try:
+            return super().fromtarfile(archive)
+        except (
+            tarfile.EmptyHeaderError,  # type: ignore[attr-defined]
+            tarfile.TruncatedHeaderError,  # type: ignore[attr-defined]
+            tarfile.InvalidHeaderError,  # type: ignore[attr-defined]
+        ) as exc:
+            # TarFile otherwise treats a short/invalid header after its first
+            # member as an ordinary end of archive, hiding the truncation.
+            raise tarfile.ReadError(str(exc)) from exc
+
+
 class CmdIOSCheckSysdiagnose(Command):
     def __init__(
         self,
@@ -63,6 +92,8 @@ class CmdIOSCheckSysdiagnose(Command):
         self.ips_files: list[dict[str, Any]] = []
         self.temp_sysdiagnose_dir: Optional[TemporaryDirectory[str]] = None
         self.extracted_sysdiagnose_path: Optional[str] = None
+        self.sysdiagnose_tar_members: list[tarfile.TarInfo] = []
+        self.archive_incomplete = False
 
     @staticmethod
     def _parse_bugtype_header(data: bytes) -> Optional[int]:
@@ -104,19 +135,40 @@ class CmdIOSCheckSysdiagnose(Command):
         self.log.info("Parsing sysdiagnose archive. This might take a while...")
         self.sysdiagnose_format = "tar"
         try:
-            self.sysdiagnose_archive = tarfile.open(self.target_path, "r:gz")
-            self._extract_sysdiagnose_archive()
+            with _StreamingGzipFile(self.target_path, "rb") as compressed:
+                self.sysdiagnose_archive = tarfile.open(
+                    fileobj=compressed, mode="r|", tarinfo=_SysdiagnoseTarInfo
+                )
+                self._extract_sysdiagnose_archive()
+                # Tar stops at its end marker, which can precede the gzip
+                # trailer. Drain the compressed stream to check its EOF/CRC.
+                if not self.archive_incomplete:
+                    try:
+                        while compressed.read1(1024 * 1024):
+                            pass
+                    except (EOFError, gzip.BadGzipFile, zlib.error) as exc:
+                        self._warn_incomplete_archive(exc)
         except (tarfile.ReadError, EOFError, zlib.error, OSError) as exc:
-            # A truncated archive ends in EOFError from gzip, which Click would
-            # otherwise report as a bare "Aborted!" with no reason.
             self.log.critical(
                 "Unable to read the sysdiagnose archive %s: %s. "
                 "The file may be truncated or not a gzip-compressed tarball.",
                 self.target_path,
                 exc,
             )
+            self.finish()
             sys.exit(1)
 
+    def _warn_incomplete_archive(self, exc: Exception) -> None:
+        self.archive_incomplete = True
+        self.log.warning(
+            "The sysdiagnose archive is truncated or damaged: %s. Recovered "
+            "%d complete files; incomplete files are excluded. Analysis will "
+            "be partial and archive integrity cannot be verified. Obtain a "
+            "complete copy for a full analysis.",
+            exc,
+            len(self.sysdiagnose_files),
+        )
+
     def _extract_sysdiagnose_archive(self) -> None:
         archive = self.sysdiagnose_archive
         if archive is None:
@@ -124,62 +176,99 @@ class CmdIOSCheckSysdiagnose(Command):
 
         self.temp_sysdiagnose_dir = TemporaryDirectory()
         extraction_root = Path(self.temp_sysdiagnose_dir.name).resolve()
-        archive_roots = set()
+        archive_roots: set[str] = set()
 
-        for member in archive:
-            member_path = PurePosixPath(member.name.replace("\\", "/"))
-            if member_path.is_absolute() or ".." in member_path.parts:
-                self.log.warning("Skipping unsafe sysdiagnose path %r", member.name)
-                continue
-
-            destination = extraction_root.joinpath(*member_path.parts).resolve()
-            if not destination.is_relative_to(extraction_root):
-                self.log.warning("Skipping unsafe sysdiagnose path %r", member.name)
-                continue
-
-            if not member_path.parts:
-                continue
-            # AppleDouble sidecars (._name) carry a file's extended attributes,
-            # not sysdiagnose content. Device archives hold hundreds of them;
-            # bsdtar hides them from listings, tarfile does not.
-            if member_path.name.startswith("._"):
-                continue
-            archive_roots.add(member_path.parts[0])
-
-            if member.isdir():
-                destination.mkdir(parents=True, exist_ok=True)
-                continue
-
-            # Modules only need directories and regular files. Do not materialize
-            # links or device nodes from an untrusted sysdiagnose archive.
-            if not member.isfile():
-                self.log.warning("Skipping unsafe sysdiagnose member %r", member.name)
-                continue
-
-            normalized_name = member_path.as_posix()
-            self.sysdiagnose_files.append(normalized_name)
-
-            source = archive.extractfile(member)
-            if source is None:
-                continue
-
-            destination.parent.mkdir(parents=True, exist_ok=True)
-            with source, destination.open("wb") as output:
-                shutil.copyfileobj(source, output)
-
-            if normalized_name.endswith(".ips"):
-                self._add_ips_file(str(destination), destination.read_bytes())
+        try:
+            for member in archive:
+                self._extract_member(archive, member, extraction_root, archive_roots)
+        except (tarfile.ReadError, EOFError, gzip.BadGzipFile, zlib.error) as exc:
+            self._warn_incomplete_archive(exc)
+            if not self.sysdiagnose_files:
+                raise tarfile.ReadError(
+                    "No complete sysdiagnose files could be recovered"
+                ) from exc
 
         if len(archive_roots) != 1:
+            self.finish()
             raise ValueError("Sysdiagnose archive must contain one top-level directory")
 
         self.extracted_sysdiagnose_path = str(extraction_root / archive_roots.pop())
 
+    def _extract_member(
+        self,
+        archive: tarfile.TarFile,
+        member: tarfile.TarInfo,
+        extraction_root: Path,
+        archive_roots: set[str],
+    ) -> None:
+        member_path = PurePosixPath(member.name.replace("\\", "/"))
+        if member_path.is_absolute() or ".." in member_path.parts:
+            self.log.warning("Skipping unsafe sysdiagnose path %r", member.name)
+            return
+
+        destination = extraction_root.joinpath(*member_path.parts).resolve()
+        if not destination.is_relative_to(extraction_root):
+            self.log.warning("Skipping unsafe sysdiagnose path %r", member.name)
+            return
+
+        if not member_path.parts:
+            return
+        # AppleDouble sidecars (._name) carry a file's extended attributes,
+        # not sysdiagnose content. Device archives hold hundreds of them;
+        # bsdtar hides them from listings, tarfile does not.
+        if member_path.name.startswith("._"):
+            return
+        archive_roots.add(member_path.parts[0])
+
+        if member.isdir():
+            destination.mkdir(parents=True, exist_ok=True)
+            self.sysdiagnose_tar_members.append(member)
+            return
+
+        # Modules only need directories and regular files. Do not materialize
+        # links or device nodes from an untrusted sysdiagnose archive.
+        if not member.isfile():
+            self.log.warning("Skipping unsafe sysdiagnose member %r", member.name)
+            return
+
+        normalized_name = member_path.as_posix()
+
+        source = archive.extractfile(member)
+        if source is None:
+            return
+
+        destination.parent.mkdir(parents=True, exist_ok=True)
+        # Stage each file independently. A truncated member must never be
+        # indexed or left for a module that scans the extracted directory.
+        # Staging also preserves an earlier complete entry with the same name.
+        with NamedTemporaryFile(dir=destination.parent, delete=False) as output:
+            staged_path = Path(output.name)
+            try:
+                with source:
+                    shutil.copyfileobj(source, output)
+                if output.tell() != member.size:
+                    raise tarfile.ReadError("Unexpected end of sysdiagnose member")
+            except BaseException:
+                output.close()
+                staged_path.unlink(missing_ok=True)
+                raise
+        try:
+            staged_path.replace(destination)
+        finally:
+            staged_path.unlink(missing_ok=True)
+        self.sysdiagnose_files.append(normalized_name)
+        self.sysdiagnose_tar_members.append(member)
+
+        if normalized_name.endswith(".ips"):
+            self._add_ips_file(str(destination), destination.read_bytes())
+
     def module_init(self, module) -> None:
         module.ips_files = self.ips_files
         if self.sysdiagnose_format == "tar":
             if self.extracted_sysdiagnose_path is None:
                 raise RuntimeError("Sysdiagnose archive has not been extracted")
+            module.sysdiagnose_tar_members = self.sysdiagnose_tar_members
+            module.sysdiagnose_archive_incomplete = self.archive_incomplete
             module.from_sysdiagnose_folder(
                 self.extracted_sysdiagnose_path, self.sysdiagnose_files
             )

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.10.0.
+    internals). Pinned to iphone_backup_decrypt==0.9.0.
     """
 
     def __init__(self, *, backup_directory, passphrase=None, derived_key=None):

src/mvt/ios/modules/fs/filesystem.py

@@ -82,9 +82,7 @@ class Filesystem(IOSExtraction):
                 try:
                     dir_path = os.path.join(root, dir_name)
                     result = {
-                        "path": os.path.relpath(dir_path, self.target_path).replace(
-                            os.sep, "/"
-                        ),
+                        "path": os.path.relpath(dir_path, self.target_path),
                         "modified": convert_unix_to_iso(os.stat(dir_path).st_mtime),
                     }
                 except Exception:
@@ -96,9 +94,7 @@ class Filesystem(IOSExtraction):
                 try:
                     file_path = os.path.join(root, file_name)
                     result = {
-                        "path": os.path.relpath(file_path, self.target_path).replace(
-                            os.sep, "/"
-                        ),
+                        "path": os.path.relpath(file_path, self.target_path),
                         "modified": convert_unix_to_iso(os.stat(file_path).st_mtime),
                     }
                 except Exception:

src/mvt/ios/modules/sysdiagnose/base.py

@@ -40,6 +40,10 @@ class SysdiagnoseExtraction(MVTModule):
         self.tar: Optional[tarfile.TarFile] = None
         self.tar_files: list[str] = []
         self.ips_files: list[dict[str, object]] = []
+        # Metadata for safely extracted, complete archive members. None for
+        # directory input or commands that do not supply archive metadata.
+        self.sysdiagnose_tar_members: Optional[list[tarfile.TarInfo]] = None
+        self.sysdiagnose_archive_incomplete = False
 
     def from_sysdiagnose_folder(
         self, target_path: str, sysdiagnose_files: list[str]
@@ -56,17 +60,21 @@ class SysdiagnoseExtraction(MVTModule):
     def _extract_timezone(self):
         """Determine the sysdiagnose timezone from its diagnostic log."""
         file_paths = self._get_files_by_pattern("*/sysdiagnose.log")
-        if not file_paths:
-            self.log.info(
-                "Unable to determine the timezone in which the sysdiagnose was "
-                "generated. Assuming UTC for logs without a timezone."
+        filenames = []
+        if file_paths:
+            content = self._get_file_content(file_paths[0]).decode(
+                "utf-8", errors="replace"
             )
-            return timezone.utc
-
-        content = self._get_file_content(file_paths[0]).decode(
-            "utf-8", errors="replace"
-        )
-        filenames = re.findall(r"sysdiagnose_\S+?\.tar\.gz", content)
+            filenames = re.findall(r"sysdiagnose_\S+?\.tar\.gz", content)
+        if not filenames and self.target_path:
+            # A recovered archive can be missing sysdiagnose.log. Its original
+            # filename still records the collection offset, when not renamed.
+            filename = Path(self.target_path).name
+            if re.match(
+                r"sysdiagnose_\d{4}\.\d{2}\.\d{2}_\d{2}-\d{2}-\d{2}[+-]\d{4}(?:_|\.)",
+                filename,
+            ):
+                filenames = [filename]
         if not filenames:
             self.log.info(
                 "Unable to determine the timezone in which the sysdiagnose was "
@@ -74,9 +82,7 @@ class SysdiagnoseExtraction(MVTModule):
             )
             return timezone.utc
 
-        timestamp = "_".join(
-            filenames[0].removesuffix(".tar.gz").split("_")[1:3]
-        )
+        timestamp = "_".join(filenames[0].removesuffix(".tar.gz").split("_")[1:3])
         sysdiagnose_timezone = datetime.strptime(
             timestamp, "%Y.%m.%d_%H-%M-%S%z"
         ).tzinfo

tests/android/test_artifact_dumpsys_accessibility.py

@@ -35,33 +35,11 @@ class TestDumpsysAccessibilityArtifact:
 
         assert len(da.results) == 0
         da.parse(data)
-        # One named service, plus one count-only record: the dump states
-        # `installedServiceCount=2` and names only one component.
-        assert len(da.results) == 2
+        assert len(da.results) == 1
         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
-        # 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_parsing_installed_and_enabled(self):
-        da = DumpsysAccessibilityArtifact()
-        with open(get_artifact("android_data/dumpsys_accessibility_enabled.txt")) as f:
-            da.parse(f.read())
-
-        assert len(da.results) == 5
-        assert all(result["component"] for result in da.results)
-        assert all(result["installed"] is True for result in da.results)
-        enabled = [result for result in da.results if result["enabled"] is True]
-        assert len(enabled) == 1
-        assert enabled[0]["package_name"] == "com.samsung.accessibility"
-        assert (
-            enabled[0]["component"]
-            == "com.samsung.accessibility/.universalswitch.UniversalSwitchService"
-        )
-        assert sum(result["enabled"] is False for result in da.results) == 4
+        assert da.results[0]["installed"] is False
 
     def test_accessibility_service_alert(self):
         da = DumpsysAccessibilityArtifact()
@@ -72,11 +50,9 @@ class TestDumpsysAccessibilityArtifact:
 
         da.check_indicators()
 
-        assert len(da.alertstore.alerts) == 2
+        assert len(da.alertstore.alerts) == 1
         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()
@@ -108,11 +84,7 @@ User state[attributes:{id=10
         assert len(da.alertstore.alerts) == 0
         da.check_indicators()
         assert len(da.alertstore.alerts) == len(da.results)
-        # 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.MEDIUM) == 3
         assert da.alertstore.count(AlertLevel.CRITICAL) == 1
         critical_alert = next(
             alert

tests/android/test_artifact_dumpsys_accessibility_count.py

@@ -1,149 +0,0 @@
-# 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

@@ -1,155 +0,0 @@
-# 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,101 +130,6 @@ 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,18 +144,3 @@ 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

@@ -1,43 +0,0 @@
-# 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_dumpsys_accessibility.py

@@ -1,73 +0,0 @@
-# Mobile Verification Toolkit (MVT)
-# Copyright (c) 2021-2026 The MVT Authors.
-# Use of this software is governed by the MVT License 1.1 that can be found at
-#   https://license.mvt.re/1.1/
-
-import logging
-
-from mvt.android.modules.bugreport.dumpsys_accessibility import DumpsysAccessibility
-
-from ..utils import get_artifact
-
-
-def _run(content, monkeypatch, caplog):
-    module = DumpsysAccessibility()
-    dumpstate = f"DUMP OF SERVICE accessibility:\n{content}".encode()
-    monkeypatch.setattr(module, "_get_dumpstate_file", lambda: dumpstate)
-    with caplog.at_level(logging.INFO):
-        module.run()
-
-
-def test_logs_mixed_enabled_states(monkeypatch, caplog):
-    with open(get_artifact("android_data/dumpsys_accessibility_enabled.txt")) as handle:
-        _run(handle.read(), monkeypatch, caplog)
-
-    assert (
-        'Found accessibility service "com.samsung.accessibility/'
-        '.universalswitch.UniversalSwitchService" (enabled: True)' in caplog.messages
-    )
-    assert sum("(enabled: False)" in message for message in caplog.messages) == 4
-    assert (
-        "Identified a total of 5 accessibility services, 1 reported enabled "
-        "(0 more stated by the dump without a component name)" in caplog.messages
-    )
-
-
-def test_logs_enabled_count_alongside_unnamed_count(monkeypatch, caplog):
-    with open(
-        get_artifact("android_data/dumpsys_accessibility_v14_or_later.txt")
-    ) as handle:
-        _run(handle.read(), monkeypatch, caplog)
-
-    assert (
-        "Identified a total of 1 accessibility services, 1 reported enabled "
-        "(1 more stated by the dump without a component name)" in caplog.messages
-    )
-    assert (
-        sum("Found accessibility service" in message for message in caplog.messages)
-        == 1
-    )
-
-
-def test_logs_unknown_enabled_state_without_treating_it_as_disabled(
-    monkeypatch, caplog
-):
-    _run(
-        """User state[attributes:{id=0, installedServiceCount=2}
-  installed services: {
-    0 : com.example/.Service
-  }
-""",
-        monkeypatch,
-        caplog,
-    )
-
-    assert (
-        'Found accessibility service "com.example/.Service" (enabled: not stated)'
-        in caplog.messages
-    )
-    assert not any("(enabled: False)" in message for message in caplog.messages)
-    assert (
-        "Identified a total of 1 accessibility services, 0 reported enabled "
-        "(1 more stated by the dump without a component name)" in caplog.messages
-    )

tests/android/test_bugreport_section_extraction.py

@@ -1,42 +0,0 @@
-# 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

@@ -1,56 +0,0 @@
-# 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/artifacts/android_data/dumpsys_accessibility_enabled.txt

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

tests/common/test_cli_plugins.py

@@ -1,4 +1,3 @@
-import shlex
 from types import SimpleNamespace
 
 import click
@@ -111,7 +110,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 {shlex.quote(str(command_path))} he"
+    words = f"group --load-command {command_path} he"
 
     result = CliRunner().invoke(
         group,

tests/common/test_cmd_plugins.py

@@ -1,5 +1,4 @@
 import json
-import re
 from types import SimpleNamespace
 
 import pytest
@@ -55,23 +54,20 @@ def _run(command, arguments):
     return CliRunner().invoke(command, arguments, env={"COLUMNS": "200"})
 
 
-def _table_lines(output):
-    """Return the cells of each line of the table, header first."""
+def _table_rows(output):
+    """Return the content of the table rows, without the header and the box."""
     return [
-        [cell.strip() for cell in re.split("[│┃]", line.strip().strip("│┃"))]
+        [cell.strip() for cell in line.strip().strip("│").split("│")]
         for line in output.splitlines()
-        if "│" in line or "┃" in line
+        if "│" 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):
-    lines = _table_lines(output)
-    return lines[0] if lines else []
+    for line in output.splitlines():
+        if "┃" in line:
+            return [cell.strip() for cell in line.strip().strip("┃").split("┃")]
+    return []
 
 
 def _install(monkeypatch, distributions, entry_points):

tests/common/test_indicators.py

@@ -311,31 +311,3 @@ 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,23 +22,3 @@ 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,7 +7,6 @@ import logging
 import threading
 from pathlib import Path
 
-import pytest
 from Crypto.Cipher import AES
 
 from mvt.ios.decrypt import DecryptBackup, MVTEncryptedBackup
@@ -82,10 +81,7 @@ def test_extract_file_by_id_copies_unencrypted_files(mocker, tmp_path):
     assert output_path.read_bytes() == b"plain content"
 
 
-@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
-):
+def test_process_backup_rejects_unsafe_file_ids_and_destinations(mocker, tmp_path):
     backup_path = tmp_path / "backup"
     destination = tmp_path / "destination"
     outside = tmp_path / "outside"
@@ -96,27 +92,20 @@ def test_process_backup_rejects_unsafe_file_ids_and_destinations(
     safe_file_id = "ef" + "3" * 38
     unsafe_file_id = "../../outside-file"
     symlink_file_id = "ab" + "4" * 38
-    file_ids = [safe_file_id]
-    if with_symlink:
-        file_ids.append(symlink_file_id)
-    for file_id in file_ids:
+    for file_id in (safe_file_id, symlink_file_id):
         source_path = backup_path / file_id[:2] / file_id
         source_path.parent.mkdir(parents=True, exist_ok=True)
         source_path.write_bytes(b"encrypted")
-    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")
+    (destination / "ab").symlink_to(outside, target_is_directory=True)
 
     cursor = mocker.MagicMock()
-    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.__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"),
+        ]
+    )
     cursor_context = mocker.MagicMock()
     cursor_context.__enter__.return_value = cursor
 
@@ -135,8 +124,7 @@ def test_process_backup_rejects_unsafe_file_ids_and_destinations(
     decryptor.process_backup()
 
     assert (destination / safe_file_id[:2] / safe_file_id).read_bytes() == b"decrypted"
-    if with_symlink:
-        assert not (outside / symlink_file_id).exists()
+    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,8 +9,6 @@ import plistlib
 import shutil
 import sqlite3
 
-import pytest
-
 from mvt.ios.modules.base import IOSExtraction
 from mvt.ios.modules.fs.analytics import Analytics
 
@@ -47,9 +45,6 @@ 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,45 +3,15 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 import logging
-import ntpath
-from types import SimpleNamespace
 
 from mvt.common.indicators import Indicators
 from mvt.common.module import run_module
-from mvt.ios.modules.fs import filesystem
 from mvt.ios.modules.fs.filesystem import Filesystem
 
 from ..utils import get_ios_backup_folder
 
 
 class TestFilesystem:
-    def test_windows_paths_are_normalized_for_indicators(
-        self, monkeypatch, indicators_factory
-    ):
-        m = Filesystem(target_path=r"C:\dump")
-        m.indicators = indicators_factory(
-            file_paths=["matched/directory"], processes=["matched"]
-        )
-        monkeypatch.setattr(
-            filesystem,
-            "os",
-            SimpleNamespace(
-                sep="\\",
-                path=ntpath,
-                walk=lambda _: [(r"C:\dump\matched", ["directory"], ["file.txt"])],
-                stat=lambda _: SimpleNamespace(st_mtime=0),
-            ),
-        )
-
-        m.run()
-        m.check_indicators()
-
-        assert {result["path"] for result in m.results} == {
-            "matched/directory",
-            "matched/file.txt",
-        }
-        assert len(m.alertstore.alerts) == 3
-
     def test_filesystem(self):
         m = Filesystem(target_path=get_ios_backup_folder())
         run_module(m)

tests/test_check_android_androidqf.py

@@ -85,7 +85,7 @@ class TestCheckAndroidqfCommand:
     def test_acquisition_context_falls_back_to_public_key_file(self, tmp_path):
         data_path = tmp_path / "androidqf"
         data_path.mkdir()
-        (data_path / "adb_host_key.pub").write_bytes(b"QUJDRA== acquisition@host\n")
+        (data_path / "adb_host_key.pub").write_text("QUJDRA== acquisition@host\n")
         command = CmdAndroidCheckAndroidQF(target_path=str(data_path))
 
         command.init()

tests/test_check_ios_sysdiagnose.py

@@ -1,4 +1,5 @@
 import logging
+import json
 import os
 import tarfile
 
@@ -6,6 +7,8 @@ from click.testing import CliRunner
 
 from mvt.ios.cli import check_sysdiagnose
 
+from .test_cmd_check_sysdiagnose import _recovery_archive
+
 
 CUSTOM_MODULE = """
 from mvt.ios.modules.sysdiagnose import SysdiagnoseExtraction
@@ -90,3 +93,27 @@ def test_check_sysdiagnose_reports_a_truncated_archive(tmp_path, caplog):
     assert "Unable to read the sysdiagnose archive" in caplog.text
     assert "truncated" in caplog.text
     assert "Aborted!" not in result.output
+
+
+def test_check_sysdiagnose_recovers_and_saves_warning(tmp_path, caplog):
+    archive = _recovery_archive(tmp_path, "gzip-body")
+    module_path = tmp_path / "custom_sysdiagnose.py"
+    module_path.write_text(CUSTOM_MODULE, encoding="utf-8")
+    output_path = tmp_path / "output"
+    with caplog.at_level(logging.WARNING, logger="mvt"):
+        result = CliRunner().invoke(
+            check_sysdiagnose,
+            [
+                "--load-module",
+                str(module_path),
+                "--output",
+                str(output_path),
+                str(archive),
+            ],
+        )
+    assert result.exit_code == 0, result.output
+    assert json.loads((output_path / "custom_sysdiagnose_module.json").read_text()) == [
+        {"content": "artifact"}
+    ]
+    assert "truncated or damaged" in caplog.text
+    assert "Recovered 3 complete files" in (output_path / "command.log").read_text()

tests/test_cmd_check_sysdiagnose.py

@@ -1,8 +1,12 @@
 import io
+import gzip
+import logging
+import random
 import tarfile
 from datetime import timedelta
 from pathlib import Path
 
+import pytest
 from mvt.ios.cmd_check_sysdiagnose import CmdIOSCheckSysdiagnose
 from mvt.ios.modules.sysdiagnose import SysdiagnoseExtraction
 
@@ -86,6 +90,8 @@ def test_check_sysdiagnose_from_archive_closes_archive(tmp_path):
     ]
     assert command.sysdiagnose_archive is None
     assert "sysdiagnose/._artifact.txt" not in command.sysdiagnose_files
+    assert command.archive_incomplete is False
+    assert _test_module(command).sysdiagnose_archive_incomplete is False
 
 
 def test_archive_is_extracted_once_and_unsafe_members_are_skipped(tmp_path):
@@ -116,8 +122,165 @@ 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 Path(module.parent_path) == extracted_path.parent
+        assert module.parent_path == str(extracted_path.parent)
     finally:
         command.finish()
 
     assert not extracted_path.exists()
+
+
+def _recovery_archive(tmp_path, damage, last_name="partial.ips"):
+    raw = io.BytesIO()
+    with tarfile.open(fileobj=raw, mode="w") as archive:
+        for name, data in (
+            ("artifact.txt", b"artifact"),
+            ("sysdiagnose.log", b"sysdiagnose_2024.01.02_03-04-05+0200.tar.gz"),
+            ("complete.ips", b'{"bug_type": 210}\ncomplete'),
+            (last_name, b'{"bug_type": 210}\n' + random.Random(42).randbytes(200_000)),
+        ):
+            member = tarfile.TarInfo(f"sysdiagnose/{name}")
+            member.size = len(data)
+            member.mtime = 1_700_000_000
+            member.pax_headers = {"LIBARCHIVE.creationtime": "1600000000.5"}
+            last_header = archive.offset
+            archive.addfile(member, io.BytesIO(data))
+    data = gzip.compress(raw.getvalue())
+    if damage == "gzip-body":
+        data = data[: len(data) // 2]
+    elif damage == "tar-body":
+        data = gzip.compress(raw.getvalue()[: last_header + 2000])
+    elif damage == "tar-header":
+        data = gzip.compress(raw.getvalue()[: last_header + 100])
+    elif damage == "gzip-trailer":
+        data = data[:-8]
+    elif damage == "checksum":
+        data = data[:-8] + bytes(x ^ 0xFF for x in data[-8:-4]) + data[-4:]
+    path = tmp_path / "sysdiagnose.tar.gz"
+    path.write_bytes(data)
+    return path
+
+
+@pytest.mark.parametrize("damage", ["gzip-body", "tar-body", "tar-header"])
+def test_recovers_complete_members_and_excludes_partial_files(tmp_path, caplog, damage):
+    path = _recovery_archive(tmp_path, damage)
+    original = path.read_bytes()
+    command = CmdIOSCheckSysdiagnose(target_path=str(path))
+    try:
+        with caplog.at_level(logging.WARNING):
+            command.init()
+        root = Path(command.extracted_sysdiagnose_path)
+        assert {p.name for p in root.iterdir()} == {
+            "artifact.txt",
+            "sysdiagnose.log",
+            "complete.ips",
+        }
+        assert (root / "artifact.txt").read_bytes() == b"artifact"
+        assert command.sysdiagnose_files == [
+            "sysdiagnose/artifact.txt",
+            "sysdiagnose/sysdiagnose.log",
+            "sysdiagnose/complete.ips",
+        ]
+        assert [r["file_path"] for r in command.ips_files] == [
+            str(root / "complete.ips")
+        ]
+        assert [
+            m.name for m in command.sysdiagnose_tar_members
+        ] == command.sysdiagnose_files
+        assert (
+            command.sysdiagnose_tar_members[0].pax_headers["LIBARCHIVE.creationtime"]
+            == "1600000000.5"
+        )
+        assert command.archive_incomplete
+        assert "truncated or damaged" in caplog.text
+        assert "Recovered 3 complete files" in caplog.text
+        assert "Analysis will be partial" in caplog.text
+        assert path.read_bytes() == original
+    finally:
+        command.finish()
+    assert not root.exists()
+
+
+@pytest.mark.parametrize("damage", ["gzip-trailer", "checksum"])
+def test_warns_when_tar_is_complete_but_gzip_integrity_fails(tmp_path, caplog, damage):
+    path = _recovery_archive(tmp_path, damage)
+    with caplog.at_level(logging.WARNING):
+        command = _run_command(path)
+    assert _test_module(command).results[0]["content"] == "artifact"
+    assert command.archive_incomplete
+    assert _test_module(command).sysdiagnose_archive_incomplete
+    assert len(command.sysdiagnose_files) == 4
+    assert "archive integrity cannot be verified" in caplog.text
+
+
+def test_truncated_duplicate_preserves_earlier_complete_file(tmp_path):
+    path = _recovery_archive(tmp_path, "gzip-body", last_name="artifact.txt")
+    command = _run_command(path)
+    assert _test_module(command).results[0]["content"] == "artifact"
+    assert command.sysdiagnose_files.count("sysdiagnose/artifact.txt") == 1
+
+
+def test_output_write_error_is_fatal_not_recovered(tmp_path, monkeypatch, caplog):
+    path = _recovery_archive(tmp_path, "none")
+    command = CmdIOSCheckSysdiagnose(target_path=str(path))
+
+    def fail_copy(*args):
+        raise OSError("disk full")
+
+    monkeypatch.setattr("mvt.ios.cmd_check_sysdiagnose.shutil.copyfileobj", fail_copy)
+    with pytest.raises(SystemExit) as error:
+        command.init()
+    assert error.value.code == 1
+    assert not command.archive_incomplete
+    assert "disk full" in caplog.text
+    assert command.temp_sysdiagnose_dir is None
+    assert command.sysdiagnose_archive is None
+
+
+@pytest.mark.parametrize("offset", ["+0200", "-0530"])
+def test_recovered_timezone_falls_back_to_original_filename(tmp_path, offset):
+    path = _recovery_archive(tmp_path, "gzip-body")
+    renamed = path.with_name(
+        f"sysdiagnose_2026.09.14_20-04-52{offset}_iPhone-OS_iPhone_24A000.tar.gz"
+    )
+    path.rename(renamed)
+    command = CmdIOSCheckSysdiagnose(target_path=str(renamed))
+    try:
+        command.init()
+        module = SysdiagnoseExtraction(target_path=str(renamed))
+        command.module_init(module)
+        # Simulate the diagnostic log being in the missing part of an archive.
+        module.files = [f for f in module.files if not f.endswith("/sysdiagnose.log")]
+        expected = (
+            timedelta(hours=2) if offset == "+0200" else -timedelta(hours=5, minutes=30)
+        )
+        assert module._extract_timezone().utcoffset(None) == expected
+    finally:
+        command.finish()
+
+
+def test_recovery_still_skips_unsafe_paths_and_links(tmp_path):
+    raw = io.BytesIO()
+    with tarfile.open(fileobj=raw, mode="w") as archive:
+        for name in ("sysdiagnose/artifact.txt", "sysdiagnose/../../escaped.txt"):
+            member = tarfile.TarInfo(name)
+            member.size = 8
+            archive.addfile(member, io.BytesIO(b"artifact"))
+        link = tarfile.TarInfo("sysdiagnose/link")
+        link.type = tarfile.SYMTYPE
+        link.linkname = "/etc/hostname"
+        archive.addfile(link)
+    path = tmp_path / "sysdiagnose.tar.gz"
+    path.write_bytes(gzip.compress(raw.getvalue())[:-8])
+    command = CmdIOSCheckSysdiagnose(target_path=str(path))
+    try:
+        command.init()
+        root = Path(command.extracted_sysdiagnose_path)
+        assert {p.name for p in root.iterdir()} == {"artifact.txt"}
+        assert command.sysdiagnose_files == ["sysdiagnose/artifact.txt"]
+        assert [
+            m.name for m in command.sysdiagnose_tar_members
+        ] == command.sysdiagnose_files
+        assert not (root.parent.parent / "escaped.txt").exists()
+        assert command.archive_incomplete
+    finally:
+        command.finish()

tests/test_completion.py

@@ -43,7 +43,6 @@ 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"])
@@ -68,7 +67,6 @@ 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,6 +6,22 @@ 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"
@@ -609,14 +625,14 @@ wheels = [
 
 [[package]]
 name = "iphone-backup-decrypt"
-version = "0.10.0"
+version = "0.9.0"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "pycryptodome" },
 ]
-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" }
+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" }
 wheels = [
-    { 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" },
+    { 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" },
 ]
 
 [[package]]
@@ -761,6 +777,17 @@ 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"
@@ -989,11 +1016,13 @@ 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" },
@@ -1034,11 +1063,13 @@ 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.10.0" },
+    { name = "iphone-backup-decrypt", specifier = "==0.9.0" },
+    { name = "libusb1", specifier = "==3.4.0" },
     { name = "nskeyedunarchiver", specifier = "==1.5.2" },
     { name = "packaging", specifier = "==26.3" },
     { name = "pyahocorasick", specifier = "==2.3.1" },
@@ -1243,6 +1274,15 @@ 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"
@@ -1644,6 +1684,18 @@ 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"