Merge branch 'main' into feature/installed-module-packages

Modules loaded from installed packages or file paths live outside the
mvt logger hierarchy, so their log records never reach MVT's console
and file handlers and instead fall through to logging.lastResort:
alerts print as bare unformatted lines and INFO messages are dropped
entirely.

Add get_module_logger() and use it everywhere module loggers are
created. Built-in mvt.* modules keep their existing logger names, and
everything external is parented under a dedicated mvt.ext namespace so
records reach the handlers and external names can never collide with
MVT's internal logger tree. File-path modules are named after their
file (mvt.ext.<stem>) instead of the mangled internal import name.

Document a naming convention for community module packages:
distribute as mvt-plugin-<name> with import package mvt_plugin_<name>,
including the publishing organization in the name. The prefix is
advisory (loading is by entry point, and it is no mark of
authenticity), but conforming packages get a cleaner logger namespace:
the mvt_plugin_ prefix is stripped, so mvt_plugin_amnesty_custom logs
as mvt.ext.amnesty_custom.

src/mvt/android/artifacts/dumpsys_accessibility.py

@@ -4,7 +4,6 @@
 #   https://license.mvt.re/1.1/
 
 import re
-from typing import Any
 
 from .artifact import AndroidArtifact
 
@@ -21,10 +20,10 @@ class DumpsysAccessibilityArtifact(AndroidArtifact):
                     continue
 
             self.alertstore.medium(
-                f'Found accessibility service: "{result["component"]}"',
+                f'Found accessibility service: "{result["service"]}"',
                 "",
                 result,
-            )
+                )
 
     def parse(self, content: str) -> None:
         """
@@ -34,69 +33,41 @@ class DumpsysAccessibilityArtifact(AndroidArtifact):
         :param content: content of the accessibility section (string)
         """
 
-        self.results: list[dict[str, Any]] = []
-        services: dict[tuple[int | None, str], dict] = {}
-        user_id: int | None = None
-        state: str | None = None
+        # "Old" syntax
+        in_services = False
+        for line in content.splitlines():
+            if line.strip().startswith("installed services:"):
+                in_services = True
+                continue
+
+            if not in_services:
+                continue
+
+            if line.strip() == "}":
+                # At end of installed services
+                break
+
+            service = line.split(":")[1].strip()
+
+            self.results.append(
+                {
+                    "package_name": service.split("/")[0],
+                    "service": service,
+                }
+            )
+
+        # "New" syntax - AOSP >= 14 (?)
+        # Looks like:
+        # Enabled services:{{com.azure.authenticator/com.microsoft.brooklyn.module.accessibility.BrooklynAccessibilityService}, {com.agilebits.onepassword/com.agilebits.onepassword.filling.accessibility.FillingAccessibilityService}}
 
         for line in content.splitlines():
-            user_match = re.search(r"attributes:\{id=(\d+)", line)
-            if user_match:
-                user_id = int(user_match.group(1))
+            if line.strip().startswith("Enabled services:"):
+                matches = re.finditer(r"{([^{]+?)}", line)
 
-            stripped = line.strip()
-            state_match = re.match(
-                r"(?i)(installed|enabled|binding|bound|crashed) services\s*:\s*\{(.*)",
-                stripped,
-            )
-            if state_match:
-                state = state_match.group(1).lower()
-                inline = state_match.group(2)
-                for component in re.findall(
-                    r"\{?([\w.$-]+/[\w.$-]+)(?:\s+\(A11yTool\))?\}?", inline
-                ):
-                    service = services.setdefault(
-                        (user_id, component), self._new_service(component, user_id)
+                for match in matches:
+                    # Each match is in format: <package_name>/<service>
+                    package_name, _, service = match.group(1).partition("/")
+
+                    self.results.append(
+                        {"package_name": package_name, "service": service}
                     )
-                    service[self._state_field(state)] = True
-                    service["accessibility_tool"] = "(A11yTool)" in inline
-                continue
-
-            if not state:
-                continue
-            if stripped == "}" or stripped.startswith("AccessibilityInputFilter"):
-                state = None
-                continue
-            component_match = re.search(
-                r"(?:\d+\s*:\s*)?([\w.$-]+/[\w.$-]+)(?:\s+\(A11yTool\))?",
-                stripped,
-            )
-            if component_match:
-                component = component_match.group(1)
-                service = services.setdefault(
-                    (user_id, component), self._new_service(component, user_id)
-                )
-                service[self._state_field(state)] = True
-                service["accessibility_tool"] = "(A11yTool)" in stripped
-
-        self.results.extend(services.values())
-
-    @staticmethod
-    def _state_field(state: str) -> str:
-        return {"binding": "binding", "bound": "bound"}.get(state, state)
-
-    @staticmethod
-    def _new_service(component: str, user_id: int | None) -> dict:
-        package_name, service_name = component.split("/", 1)
-        return {
-            "user_id": user_id,
-            "component": component,
-            "package_name": package_name,
-            "service_name": service_name,
-            "installed": False,
-            "enabled": False,
-            "binding": False,
-            "bound": False,
-            "crashed": False,
-            "accessibility_tool": False,
-        }

src/mvt/android/artifacts/dumpsys_adb.py

@@ -6,7 +6,6 @@
 import base64
 import binascii
 import hashlib
-import re
 
 from .artifact import AndroidArtifact
 
@@ -99,34 +98,6 @@ class DumpsysADBArtifact(AndroidArtifact):
 
         return keystore
 
-    def parse_binary_xml(self, data: bytes) -> list[dict]:
-        """Recover ADB key records from Android binary XML (ABX).
-
-        Some dumpstate implementations embed ABX in a text stream and replace
-        binary token bytes. The public key remains intact, while unavailable
-        numeric metadata is represented as ``None`` rather than corrupt text.
-        """
-        keystore = []
-        seen = set()
-        for match in re.finditer(
-            rb"(?<![A-Za-z0-9+/])([A-Za-z0-9+/]{300,}={0,2})"
-            rb"(?: ([A-Za-z0-9_.@-]+))?",
-            data,
-        ):
-            key = match.group(1)
-            try:
-                base64.b64decode(key, validate=True)
-            except (binascii.Error, ValueError):
-                continue
-            if key in seen:
-                continue
-            seen.add(key)
-            full_key = key + (b" " + match.group(2) if match.group(2) else b"")
-            key_info = self.calculate_key_info(full_key)
-            key_info["last_connected"] = None
-            keystore.append(key_info)
-        return keystore
-
     @staticmethod
     def calculate_key_info(user_key: bytes) -> dict:
         if b" " in user_key:
@@ -147,7 +118,7 @@ class DumpsysADBArtifact(AndroidArtifact):
         return {
             "user": user.decode("utf-8"),
             "fingerprint": key_fingerprint_colon,
-            "key": key_base64.decode("ascii", errors="replace"),
+            "key": key_base64,
         }
 
     def check_indicators(self) -> None:
@@ -204,23 +175,12 @@ class DumpsysADBArtifact(AndroidArtifact):
         # Keystore is in XML format on some devices and we need to parse it
         if keystore_data and keystore_data.startswith(b"<?xml"):
             parsed["debugging_manager"]["keystore"] = self.parse_xml(keystore_data)
-        elif keystore_data and keystore_data.startswith(b"ABX\x00"):
-            parsed["debugging_manager"]["keystore"] = self.parse_binary_xml(
-                keystore_data
-            )
         else:
             # Keystore is not XML format
             parsed["debugging_manager"]["keystore"] = keystore_data
 
         parsed = parsed["debugging_manager"]
 
-        for key, value in list(parsed.items()):
-            if isinstance(value, bytes):
-                decoded = value.decode("utf-8", errors="replace")
-                parsed[key] = (
-                    decoded == "true" if decoded in ("true", "false") else decoded
-                )
-
         # Calculate key fingerprints for better readability
         key_info = []
         for user_key in parsed.get("user_keys", []):

src/mvt/android/artifacts/dumpsys_appops.py

@@ -4,7 +4,6 @@
 #   https://license.mvt.re/1.1/
 
 from datetime import datetime
-import re
 from typing import Any
 
 from mvt.common.module_types import ModuleAtomicResult, ModuleSerializedResult
@@ -28,14 +27,14 @@ class DumpsysAppopsArtifact(AndroidArtifact):
                 continue
 
             for entry in perm["entries"]:
-                if entry.get("timestamp"):
+                if "timestamp" in entry:
                     records.append(
                         {
                             "timestamp": entry["timestamp"],
                             "module": self.__class__.__name__,
-                            "event": entry["event"],
+                            "event": entry["access"],
                             "data": f"{result['package_name']} access to "
-                            f"{perm['name']}: {entry['event']}",
+                            f"{perm['name']}: {entry['access']}",
                         }
                     )
 
@@ -52,7 +51,7 @@ class DumpsysAppopsArtifact(AndroidArtifact):
                     continue
 
             # We use a placeholder entry to create a basic alert even without permission entries.
-            placeholder_entry = {"event": "unknown", "timestamp": ""}
+            placeholder_entry = {"access": "Unknown", "timestamp": ""}
 
             for perm in result["permissions"]:
                 if (
@@ -61,12 +60,12 @@ class DumpsysAppopsArtifact(AndroidArtifact):
                 ):
                     for entry in sorted(
                         perm["entries"] or [placeholder_entry],
-                        key=lambda x: x.get("timestamp") or "",
+                        key=lambda x: x["timestamp"],
                     ):
                         cleaned_result = result.copy()
                         cleaned_result["permissions"] = [perm]
                         self.alertstore.medium(
-                            f"Package '{result['package_name']}' had risky permission '{perm['name']}' set to '{entry['event']}' at {entry['timestamp']}",
+                            f"Package '{result['package_name']}' had risky permission '{perm['name']}' set to '{entry['access']}' at {entry['timestamp']}",
                             entry["timestamp"],
                             cleaned_result,
                         )
@@ -74,151 +73,111 @@ class DumpsysAppopsArtifact(AndroidArtifact):
                 elif result["package_name"] in RISKY_PACKAGES:
                     for entry in sorted(
                         perm["entries"] or [placeholder_entry],
-                        key=lambda x: x.get("timestamp") or "",
+                        key=lambda x: x["timestamp"],
                     ):
                         cleaned_result = result.copy()
                         cleaned_result["permissions"] = [perm]
                         self.alertstore.medium(
-                            f"Risky package '{result['package_name']}' had '{perm['name']}' permission set to '{entry['event']}' at {entry['timestamp']}",
+                            f"Risky package '{result['package_name']}' had '{perm['name']}' permission set to '{entry['access']}' at {entry['timestamp']}",
                             entry["timestamp"],
                             cleaned_result,
                         )
 
     def parse(self, output: str) -> None:
-        self.results: list[dict[str, Any]] = []
-        permission: dict[str, Any] | None = None
-        package: dict[str, Any] | None = None
-        uid: str | None = None
-        uid_details: dict[str, Any] = {}
-        attribution: str | None = None
+        # self.results: List[Dict[str, Any]] = []
+        perm: dict[str, Any] = {}
+        package: dict[str, Any] = {}
+        entry: dict[str, Any] = {}
+        uid = None
         in_packages = False
 
-        def finish_permission() -> None:
-            nonlocal permission
-            if package is not None and permission is not None:
-                package["permissions"].append(permission)
-            permission = None
-
-        def finish_package() -> None:
-            nonlocal package
-            finish_permission()
-            if package is not None:
-                self.results.append(package)
-            package = None
-
         for line in output.splitlines():
-            uid_match = re.match(r"^  Uid ([^:]+):$", line)
-            if uid_match:
+            if line.startswith("  Uid 0:"):
                 in_packages = True
-                finish_package()
-                uid = uid_match.group(1)
-                uid_details = {
-                    "uid_state": None,
-                    "capability": None,
-                    "app_widget_visible": None,
-                    "default_modes": {},
-                }
-                continue
 
             if not in_packages:
                 continue
 
-            uid_property = re.match(
-                r"^    (state|capability|appWidgetVisible)=(.*)$", line
-            )
-            if uid_property:
-                key = {
-                    "state": "uid_state",
-                    "appWidgetVisible": "app_widget_visible",
-                }.get(uid_property.group(1), uid_property.group(1))
-                value: Any = uid_property.group(2)
-                if value in ("true", "false"):
-                    value = value == "true"
-                uid_details[key] = value
-                continue
+            if line.startswith("  Uid "):
+                uid = line[6:-1]
+                if entry:
+                    perm["entries"].append(entry)
+                    entry = {}
+                if package:
+                    if perm:
+                        package["permissions"].append(perm)
 
-            default_mode = re.match(r"^      ([A-Z0-9_]+): mode=([^\s]+)", line)
-            if default_mode and package is None:
-                uid_details["default_modes"][default_mode.group(1)] = (
-                    default_mode.group(2)
-                )
+                    perm = {}
+                    self.results.append(package)
+                package = {}
                 continue
 
             if line.startswith("    Package "):
-                finish_package()
+                if entry:
+                    perm["entries"].append(entry)
+                    entry = {}
+
+                if package:
+                    if perm:
+                        package["permissions"].append(perm)
+
+                    perm = {}
+                    self.results.append(package)
+
                 package = {
                     "package_name": line[12:-1],
                     "permissions": [],
                     "uid": uid,
-                    **uid_details,
                 }
                 continue
 
-            operation_match = re.match(
-                r"^      ([A-Z0-9_]+)(?: \(([^)]+)\))?:\s*$", line
-            )
-            if package is not None and operation_match:
-                finish_permission()
-                permission = {
-                    "name": operation_match.group(1),
-                    "mode": operation_match.group(2),
-                    "entries": [],
-                }
-                attribution = None
+            if package and line.startswith("      ") and line[6] != " ":
+                if entry:
+                    perm["entries"].append(entry)
+                    entry = {}
+                if perm:
+                    package["permissions"].append(perm)
+                    perm = {}
+
+                perm["name"] = line.split()[0]
+                perm["entries"] = []
+                if len(line.split()) > 1:
+                    perm["access"] = line.split()[1][1:-2]
+
                 continue
 
-            attribution_match = re.match(r"^\s{8,}([^=]+)=\[$", line)
-            if attribution_match:
-                attribution = attribution_match.group(1).strip()
-                continue
-            if line.strip() == "]":
-                attribution = None
-                continue
+            if line.startswith("          "):
+                # Permission entry like:
+                # Reject: [fg-s]2021-05-19 22:02:52.054 (-314d1h25m2s33ms)
+                access_type = line.split(":")[0].strip()
+                if access_type not in ["Access", "Reject"]:
+                    # Skipping invalid access type. Some entries are not in the format we expect
+                    continue
 
-            if permission is None:
-                continue
-            event_match = re.match(
-                r"^\s*(Access|Reject):\s*\[([^]]+)\]\s*"
-                r"(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d+)\s*"
-                r"(\([^)]*\))?(?:\s+duration=([^\s]+))?",
-                line,
-            )
-            running_match = re.match(
-                r"^\s*Running start at:\s*(\S+(?: \S+)?)",
-                line,
-            )
-            if event_match:
-                entry = {
-                    "event": event_match.group(1).lower(),
-                    "access": event_match.group(1),
-                    "uid_state": event_match.group(2),
-                    "timestamp": convert_datetime_to_iso(
-                        datetime.strptime(event_match.group(3), "%Y-%m-%d %H:%M:%S.%f")
-                    ),
-                    "relative_time": event_match.group(4),
-                    "duration": event_match.group(5),
-                    "attribution": attribution,
-                }
-                permission["entries"].append(entry)
-            elif running_match:
-                raw_start = running_match.group(1)
-                timestamp = None
-                if re.fullmatch(r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d+", raw_start):
-                    timestamp = convert_datetime_to_iso(
-                        datetime.strptime(raw_start, "%Y-%m-%d %H:%M:%S.%f")
+                if entry:
+                    perm["entries"].append(entry)
+                    entry = {}
+
+                entry["access"] = access_type
+                entry["type"] = line[line.find("[") + 1 : line.find("]")]
+
+                try:
+                    entry["timestamp"] = convert_datetime_to_iso(
+                        datetime.strptime(
+                            line[line.find("]") + 1 : line.find("(")].strip(),
+                            "%Y-%m-%d %H:%M:%S.%f",
+                        )
                     )
-                permission["entries"].append(
-                    {
-                        "event": "running",
-                        "access": "Running",
-                        "uid_state": None,
-                        "timestamp": timestamp,
-                        "relative_time": raw_start
-                        if raw_start.startswith("+")
-                        else None,
-                        "duration": None,
-                        "attribution": attribution,
-                    }
-                )
+                except ValueError:
+                    # Invalid date format
+                    pass
 
-        finish_package()
+            if line.strip() == "":
+                break
+
+        if entry:
+            perm["entries"].append(entry)
+        if perm:
+            package["permissions"].append(perm)
+        if package:
+            self.results.append(package)

src/mvt/android/artifacts/dumpsys_battery_daily.py

@@ -18,18 +18,18 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact):
     def serialize(self, record: ModuleAtomicResult) -> ModuleSerializedResult:
         action = record.get("action", "update")
         package_name = record["package_name"]
-        vers = record["version_code"]
+        vers = record["vers"]
 
-        if vers == 0:
+        if vers == "0":
             data = f"Recorded uninstall of package {package_name} (vers 0)"
         elif action == "downgrade":
-            prev_vers = record.get("previous_version_code", "unknown")
+            prev_vers = record.get("previous_vers", "unknown")
             data = f"Recorded downgrade of package {package_name} from vers {prev_vers} to vers {vers}"
         else:
             data = f"Recorded update of package {package_name} with vers {vers}"
 
         return {
-            "timestamp": record["period_start"],
+            "timestamp": record["from"],
             "module": self.__class__.__name__,
             "event": "battery_daily",
             "data": data,
@@ -48,7 +48,6 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact):
                 continue
 
     def parse(self, output: str) -> None:
-        self.results = []
         daily = None
         daily_updates: list[dict[str, Any]] = []
         records: list[dict[str, Any]] = []
@@ -60,10 +59,7 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact):
 
                 timeframe = line[13:].strip()
                 date_from, date_to = timeframe.strip(":").split(" to ", 1)
-                daily = {
-                    "period_start": self._format_daily_timestamp(date_from),
-                    "period_end": self._format_daily_timestamp(date_to),
-                }
+                daily = {"from": date_from[0:10], "to": date_to[0:10]}
                 continue
 
             if not daily:
@@ -74,30 +70,21 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact):
 
             line = line.strip().replace("Update ", "")
             package_name, vers = line.split(" ", 1)
-            vers_raw = vers.split("=", 1)[1]
-            try:
-                version_code: int | str = int(vers_raw)
-            except ValueError:
-                version_code = vers_raw
+            vers_nr = vers.split("=", 1)[1]
 
             already_seen = False
             for update in daily_updates:
-                if (
-                    package_name == update["package_name"]
-                    and version_code == update["version_code"]
-                ):
-                    update["occurrences"] += 1
+                if package_name == update["package_name"] and vers_nr == update["vers"]:
                     already_seen = True
                     break
 
             if not already_seen:
                 update_record: dict[str, Any] = {
                     "action": "update",
-                    "period_start": daily["period_start"],
-                    "period_end": daily["period_end"],
+                    "from": daily["from"],
+                    "to": daily["to"],
                     "package_name": package_name,
-                    "version_code": version_code,
-                    "occurrences": 1,
+                    "vers": vers_nr,
                 }
 
                 daily_updates.append(update_record)
@@ -108,31 +95,26 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact):
         self._detect_uninstalls_and_downgrades(records)
         self.results.extend(records)
 
-    @staticmethod
-    def _format_daily_timestamp(value: str) -> str:
-        if len(value) >= 19 and value[10] == "-":
-            return f"{value[:10]} {value[11:].replace('-', ':')}"
-        return value
-
-    def _detect_uninstalls_and_downgrades(self, records: list[dict[str, Any]]) -> None:
+    def _detect_uninstalls_and_downgrades(
+        self, records: list[dict[str, Any]]
+    ) -> None:
         package_versions: dict[str, int] = {}
 
         for record in sorted(
             records,
             key=lambda record: (
-                record["period_start"],
-                record["period_end"],
+                record["from"],
+                record["to"],
                 record["package_name"],
             ),
         ):
             package_name = record["package_name"]
-            vers_nr = record["version_code"]
+            vers_nr = record["vers"]
 
-            if vers_nr == 0:
-                record["action"] = "uninstall"
+            if vers_nr == "0":
                 self.alertstore.medium(
                     f"Detected uninstall of package {package_name} (vers 0)",
-                    record["period_start"],
+                    record["from"],
                     record,
                 )
                 package_versions.pop(package_name, None)
@@ -146,11 +128,11 @@ class DumpsysBatteryDailyArtifact(AndroidArtifact):
             previous_vers = package_versions.get(package_name)
             if previous_vers is not None and current_vers < previous_vers:
                 record["action"] = "downgrade"
-                record["previous_version_code"] = previous_vers
+                record["previous_vers"] = str(previous_vers)
                 self.alertstore.medium(
                     f"Detected downgrade of package {package_name} "
                     f"from vers {previous_vers} to vers {current_vers}",
-                    record["period_start"],
+                    record["from"],
                     record,
                 )
 

src/mvt/android/artifacts/dumpsys_battery_history.py

@@ -1,177 +1,96 @@
 # Mobile Verification Toolkit (MVT)
-# Copyright (c) 2021-2026 The MVT Authors.
+# 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/
 
-import datetime
-import re
-
-from mvt.common.utils import convert_datetime_to_iso
-
 from .artifact import AndroidArtifact
 
 
 class DumpsysBatteryHistoryArtifact(AndroidArtifact):
-    """Parser for package-related events in dumpsys batterystats history."""
+    """
+    Parser for dumpsys dattery history events.
+    """
 
     def check_indicators(self) -> None:
         if not self.indicators:
             return
+
         for result in self.results:
-            package_name = result.get("package_name")
-            if not package_name:
-                continue
-            ioc_match = self.indicators.check_app_id(package_name)
+            ioc_match = self.indicators.check_app_id(result["package_name"])
             if ioc_match:
                 self.alertstore.critical(
                     ioc_match.message, "", result, matched_indicator=ioc_match.ioc
                 )
-
-    @staticmethod
-    def _parse_wall_time(value: str) -> datetime.datetime | None:
-        if re.fullmatch(r"\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d+", value):
-            value = f"1900-{value}"
-        for date_format in (
-            "%Y-%m-%d-%H-%M-%S-%f",
-            "%Y-%m-%d-%H-%M-%S",
-            "%Y-%m-%d %H:%M:%S.%f",
-        ):
-            try:
-                return datetime.datetime.strptime(value, date_format)
-            except ValueError:
-                pass
-        return None
-
-    @staticmethod
-    def _elapsed_seconds(value: str) -> float | None:
-        if not value.startswith("+"):
-            return None
-        units = {"d": 86400, "h": 3600, "m": 60, "s": 1, "ms": 0.001}
-        total = 0.0
-        for number, unit in re.findall(r"(\d+)(ms|d|h|m|s)", value):
-            total += int(number) * units[unit]
-        return total
-
-    @staticmethod
-    def _package_from_name(name: str) -> str | None:
-        clean = name.removeprefix("*walarm*:").removeprefix("*alarm*:")
-        slash_parts = clean.split("/")
-        if len(slash_parts) > 1:
-            first = slash_parts[0].lstrip("@")
-            if first.startswith(("com.", "org.", "net.")):
-                return first
-            for part in reversed(slash_parts[1:]):
-                candidate = part.lstrip("@").split(":", 1)[0]
-                if candidate.startswith(("com.", "org.", "net.")):
-                    return candidate
-        parts = clean.split(".")
-        package_parts = []
-        for part in parts:
-            if part and part[0].islower():
-                package_parts.append(part)
-            else:
-                break
-        return ".".join(package_parts) if len(package_parts) >= 2 else None
-
-    @staticmethod
-    def _normalize_service(service: str) -> str:
-        # WorkManager decorates jobs with one or more scheduler prefixes.
-        if "@" in service:
-            candidates = [part for part in service.split("@") if "/" in part]
-            if candidates:
-                return candidates[-1]
-        return service
+                continue
 
     def parse(self, data: str) -> None:
-        self.results: list[dict[str, str | None]] = []
-        anchor_time: datetime.datetime | None = None
-        anchor_elapsed = 0.0
-        has_history_heading = any(
-            line.startswith("Battery History") for line in data.splitlines()
-        )
-        in_history = not has_history_heading
-
         for line in data.splitlines():
-            stripped = line.strip()
-            if line.startswith("Battery History"):
-                if in_history:
-                    break
-                in_history = True
+            if line.startswith("Battery History "):
                 continue
-            if not in_history:
-                continue
-            if has_history_heading and not stripped:
+
+            if line.strip() == "":
                 break
-            reset_match = re.search(r"(?:RESET:)?TIME:\s*(\S+)", stripped)
-            if reset_match:
-                parsed_time = self._parse_wall_time(reset_match.group(1))
-                if parsed_time is not None:
-                    elapsed_token = stripped.split()[0]
-                    anchor_elapsed = self._elapsed_seconds(elapsed_token) or 0.0
-                    anchor_time = parsed_time
-                continue
 
-            fields = stripped.split()
-            if not fields:
-                continue
-            if len(fields) > 1 and re.fullmatch(r"\d{2}-\d{2}", fields[0]):
-                time_elapsed = " ".join(fields[:2])
-                line_time = self._parse_wall_time(time_elapsed)
-                elapsed = None
+            time_parts = line.strip().split()
+            time_elapsed = time_parts[0]
+            if (
+                len(time_parts) > 1
+                and len(time_parts[0]) == 5
+                and time_parts[0][2] == "-"
+                and ":" in time_parts[1]
+            ):
+                time_elapsed = " ".join(time_parts[:2])
+
+            event = ""
+            if line.find("+job") > 0:
+                event = "start_job"
+                payload = line.split("+job=", 1)[1]
+                uid, separator, service = payload.partition(":")
+                if not separator:
+                    continue
+                service = service.strip().strip('"')
+                package_name = service.split("/")[0]
+            elif line.find("-job") > 0:
+                event = "end_job"
+                payload = line.split("-job=", 1)[1]
+                uid, separator, service = payload.partition(":")
+                if not separator:
+                    continue
+                service = service.strip().strip('"')
+                package_name = service.split("/")[0]
+            elif line.find("+running +wake_lock=") > 0:
+                payload = line.split("+running +wake_lock=", 1)[1]
+                uid, separator, _ = payload.partition(":")
+                if not separator:
+                    continue
+                event = "wake"
+                service = (
+                    line[line.find("*walarm*:") + 9 :].split(" ")[0].strip('"').strip()
+                )
+                if service == "" or "/" not in service:
+                    continue
+
+                package_name = service.split("/")[0]
+            elif (line.find("+top=") > 0) or (line.find("-top") > 0):
+                if line.find("+top=") > 0:
+                    event = "start_top"
+                    top_pos = line.find("+top=")
+                else:
+                    event = "end_top"
+                    top_pos = line.find("-top=")
+                colon_pos = top_pos + line[top_pos:].find(":")
+                uid = line[top_pos + 5 : colon_pos]
+                service = ""
+                package_name = line[colon_pos + 1 :].strip('"')
             else:
-                time_elapsed = fields[0]
-                elapsed = self._elapsed_seconds(time_elapsed)
-                line_time = None
+                continue
 
-            timestamp = line_time
-            if timestamp is None and anchor_time is not None and elapsed is not None:
-                timestamp = anchor_time + datetime.timedelta(
-                    seconds=elapsed - anchor_elapsed
-                )
-
-            def add(
-                event: str, uid: str, service: str, package_name: str | None
-            ) -> None:
-                self.results.append(
-                    {
-                        "time_elapsed": time_elapsed,
-                        "timestamp": convert_datetime_to_iso(timestamp)
-                        if timestamp
-                        else None,
-                        "event": event,
-                        "uid": uid,
-                        "package_name": package_name,
-                        "service": service,
-                    }
-                )
-
-            for sign, uid, raw_service in re.findall(
-                r"([+-])job=([^:\s]+):\"([^\"]+)\"", line
-            ):
-                service = self._normalize_service(raw_service)
-                add(
-                    "start_job" if sign == "+" else "end_job",
-                    uid,
-                    service,
-                    self._package_from_name(service) or service.split("/", 1)[0],
-                )
-
-            for sign, uid, package_name in re.findall(
-                r"([+-])top=([^:\s]+):\"([^\"]+)\"", line
-            ):
-                add(
-                    "start_top" if sign == "+" else "end_top",
-                    uid,
-                    "",
-                    package_name,
-                )
-
-            wake_match = re.search(r"\+wake_lock=([^:\s]+):\"([^\"]+)\"", line)
-            if wake_match:
-                wake_name = wake_match.group(2)
-                add(
-                    "wake",
-                    wake_match.group(1),
-                    wake_name,
-                    self._package_from_name(wake_name),
-                )
+            self.results.append(
+                {
+                    "time_elapsed": time_elapsed,
+                    "event": event,
+                    "uid": uid,
+                    "package_name": package_name,
+                    "service": service,
+                }
+            )

src/mvt/android/artifacts/dumpsys_dbinfo.py

@@ -29,27 +29,16 @@ class DumpsysDBInfoArtifact(AndroidArtifact):
 
     def parse(self, output: str) -> None:
         rxp = re.compile(
-            r"^\s*\d+:\s*\[((?:\d{4}-)?\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{3})\]\s*"
-            r"(?:\[Pid:\((\d+)\)\])?\s*([\w-]+) took (\d+)ms - ([^,]+),"
-            r"\s*sql=\"(.*)\"(?:, path=(.*))?$"
-        )
+            r".*\[((?:[0-9]{4}-)?[0-9]{2}-[0-9]{2} "
+            r"[0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{3})\]\s*"
+            r"(?:\[Pid:\((\d+)\)\])?([\w-]+).*?sql=\"(.+?)\""
+        )  # pylint: disable=line-too-long
 
-        pool: str | None = None
-        connection_number: int | None = None
-        is_primary: bool | None = None
+        pool = None
         in_operations = False
         for line in output.splitlines():
             if line.startswith("Connection pool for "):
                 pool = line.replace("Connection pool for ", "").rstrip(":")
-                in_operations = False
-
-            connection_match = re.match(r"\s+Connection #(\d+):", line)
-            if connection_match:
-                connection_number = int(connection_match.group(1))
-                is_primary = None
-
-            if line.strip().startswith("isPrimaryConnection:"):
-                is_primary = line.strip().split(":", 1)[1].strip() == "true"
 
             if not pool:
                 continue
@@ -63,6 +52,7 @@ class DumpsysDBInfoArtifact(AndroidArtifact):
 
             if not line.startswith("        "):
                 in_operations = False
+                pool = None
                 continue
 
             match = rxp.match(line)
@@ -70,15 +60,11 @@ class DumpsysDBInfoArtifact(AndroidArtifact):
                 continue
 
             result = {
-                "timestamp": match.group(1),
-                "pid": int(match.group(2)) if match.group(2) else None,
+                "isodate": match.group(1),
                 "action": match.group(3),
-                "duration_ms": int(match.group(4)),
-                "status": match.group(5),
-                "sql": match.group(6),
-                "path": match.group(7) or pool,
-                "pool_path": pool,
-                "connection_number": connection_number,
-                "is_primary": is_primary,
+                "sql": match.group(4),
+                "path": pool,
             }
+            if match.group(2):
+                result["pid"] = match.group(2)
             self.results.append(result)

src/mvt/android/artifacts/dumpsys_package_activities.py

@@ -4,7 +4,6 @@
 #   https://license.mvt.re/1.1/
 
 from .artifact import AndroidArtifact
-from .package_resolvers import parse_resolver_table
 
 
 class DumpsysPackageActivitiesArtifact(AndroidArtifact):
@@ -20,11 +19,67 @@ class DumpsysPackageActivitiesArtifact(AndroidArtifact):
                 )
                 continue
 
-    def parse(self, content: str) -> None:
+    def parse(self, content: str):
         """
         Parse the Dumpsys Package section for activities
         Adds results to self.results
 
         :param content: content of the package section (string)
         """
-        self.results = parse_resolver_table(content, "Activity")
+        self.results = []
+
+        in_activity_resolver_table = False
+        in_non_data_actions = False
+        intent = None
+        for line in content.splitlines():
+            if line.startswith("Activity Resolver Table:"):
+                in_activity_resolver_table = True
+                continue
+
+            if not in_activity_resolver_table:
+                continue
+
+            if line.startswith("  Non-Data Actions:"):
+                in_non_data_actions = True
+                continue
+
+            if not in_non_data_actions:
+                continue
+
+            # If we hit an empty line, the Non-Data Actions section should be
+            # finished.
+            if line.strip() == "":
+                break
+
+            # We detect the action name.
+            if (
+                line.startswith(" " * 6)
+                and not line.startswith(" " * 8)
+                and ":" in line
+            ):
+                intent = line.strip().replace(":", "")
+                continue
+
+            # If we are not in an intent block yet, skip.
+            if not intent:
+                continue
+
+            # If we are in a block but the line does not start with 8 spaces
+            # it means the block ended a new one started, so we reset and
+            # continue.
+            if not line.startswith(" " * 8):
+                intent = None
+                continue
+
+            # If we got this far, we are processing receivers for the
+            # activities we are interested in.
+            activity = line.strip().split(" ")[1]
+            package_name = activity.split("/")[0]
+
+            self.results.append(
+                {
+                    "intent": intent,
+                    "package_name": package_name,
+                    "activity": activity,
+                }
+            )

src/mvt/android/artifacts/dumpsys_packages.py

@@ -1,10 +1,10 @@
 # Mobile Verification Toolkit (MVT)
-# Copyright (c) 2021-2026 The MVT Authors.
+# 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/
 
 import re
-from typing import Any
+from typing import Any, Dict, List, Optional
 
 from mvt.android.utils import ROOT_PACKAGES
 from mvt.common.module_types import ModuleAtomicResult, ModuleSerializedResult
@@ -12,244 +12,222 @@ from mvt.common.module_types import ModuleAtomicResult, ModuleSerializedResult
 from .artifact import AndroidArtifact
 
 
-def _value(raw: str) -> Any:
-    if raw == "null":
-        return None
-    if raw in ("true", "false"):
-        return raw == "true"
-    try:
-        return int(raw)
-    except ValueError:
-        return raw
-
-
 class DumpsysPackagesArtifact(AndroidArtifact):
     def check_indicators(self) -> None:
         alerted_root_packages = set()
         for result in self.results:
-            package_name = result["package_name"]
-            if (
-                package_name in ROOT_PACKAGES
-                and package_name not in alerted_root_packages
-            ):
-                alerted_root_packages.add(package_name)
+            if result["package_name"] in ROOT_PACKAGES:
+                if result["package_name"] in alerted_root_packages:
+                    continue
+                alerted_root_packages.add(result["package_name"])
                 self.alertstore.medium(
-                    f'Found an installed package related to rooting/jailbreaking: "{package_name}"',
+                    f'Found an installed package related to rooting/jailbreaking: "{result["package_name"]}"',
                     "",
                     result,
                 )
+                continue
+
             if not self.indicators:
                 continue
-            ioc_match = self.indicators.check_app_id(package_name)
+
+            ioc_match = self.indicators.check_app_id(result.get("package_name", ""))
             if ioc_match:
                 self.alertstore.critical(
                     ioc_match.message, "", result, matched_indicator=ioc_match.ioc
                 )
 
     def serialize(self, record: ModuleAtomicResult) -> ModuleSerializedResult:
+        records = []
         timestamps = [
-            ("package_install", record.get("timestamp")),
-            ("package_last_update", record.get("last_update_time")),
-        ]
-        timestamps.extend(
-            ("package_first_install", user.get("first_install_time"))
-            for user in record.get("users", [])
-        )
-        return [
+            {"event": "package_install", "timestamp": record["timestamp"]},
             {
-                "timestamp": timestamp,
-                "module": self.__class__.__name__,
-                "event": event,
-                "data": f"Install or update of package {record['package_name']}",
-            }
-            for event, timestamp in timestamps
-            if timestamp
+                "event": "package_first_install",
+                "timestamp": record["first_install_time"],
+            },
+            {"event": "package_last_update", "timestamp": record["last_update_time"]},
         ]
 
+        for timestamp in timestamps:
+            records.append(
+                {
+                    "timestamp": timestamp["timestamp"],
+                    "module": self.__class__.__name__,
+                    "event": timestamp["event"],
+                    "data": f"Install or update of package {record['package_name']}",
+                }
+            )
+
+        return records
+
     @staticmethod
-    def _permission(line: str, permission_type: str) -> dict:
-        name, _, details = line.strip().partition(":")
-        granted_match = re.search(r"granted=(true|false)", details)
-        flags_match = re.search(r"flags=\[\s*([^]]*)\]", details)
-        return {
-            "name": name,
-            "type": permission_type,
-            "granted": granted_match.group(1) == "true" if granted_match else None,
-            "flags": [
-                flag.strip()
-                for flag in (flags_match.group(1).split("|") if flags_match else [])
-                if flag.strip()
-            ],
-        }
-
-    @classmethod
-    def parse_dumpsys_package_for_details(cls, output: str) -> dict[str, Any]:
-        details: dict[str, Any] = {
-            "app_id": None,
-            "version_name": None,
-            "version_code": None,
-            "min_sdk": None,
-            "target_sdk": None,
-            "timestamp": None,
-            "last_update_time": None,
-            "installer": None,
+    def parse_dumpsys_package_for_details(output: str) -> Dict[str, Any]:
+        """
+        Parse one entry of a dumpsys package information
+        """
+        details: Dict[str, Any] = {
+            "uid": "",
+            "version_name": "",
+            "version_code": "",
+            "timestamp": "",
+            "first_install_time": "",
+            "last_update_time": "",
+            "installer": "",
             "system": False,
-            "permissions": [],
-            "requested_permissions": [],
-            "users": [],
+            "permissions": list(),
+            "requested_permissions": list(),
         }
-        permission_section: str | None = None
-        current_user: dict[str, Any] | None = None
-        legacy_first_install: str | None = None
-
+        in_install_permissions = False
+        in_runtime_permissions = False
+        in_declared_permissions = False
+        in_requested_permissions = True
+        current_user: Optional[int] = None
+        first_install_times: Dict[Optional[int], str] = {}
+        runtime_permissions: Dict[Optional[int], List[Dict[str, Any]]] = {}
         for line in output.splitlines():
-            stripped = line.strip()
-            user_match = re.match(r"User (\d+):\s*(.*)", stripped)
+            user_match = re.match(r"User (\d+):", line.strip())
             if user_match:
-                current_user = {"user_id": int(user_match.group(1)), "permissions": []}
-                for key, raw in re.findall(r"(\w+)=([^\s]+)", user_match.group(2)):
-                    clean_key = {
-                        "notLaunched": "not_launched",
-                        "installReason": "install_reason",
-                        "uninstallReason": "uninstall_reason",
-                        "dataDir": "data_dir",
-                    }.get(key, re.sub(r"(?<!^)(?=[A-Z])", "_", key).lower())
-                    current_user[clean_key] = _value(raw)
-                details["users"].append(current_user)
-                permission_section = None
-                continue
+                current_user = int(user_match.group(1))
 
-            header = stripped.lower()
-            if header in {
-                "declared permissions:",
-                "install permissions:",
-                "requested permissions:",
-                "runtime permissions:",
-            }:
-                permission_section = header.split()[0]
-                continue
+            if in_install_permissions:
+                if line.startswith(" " * 4) and not line.startswith(" " * 6):
+                    in_install_permissions = False
+                else:
+                    lineinfo = line.strip().split(":")
+                    permission = lineinfo[0]
+                    granted = None
+                    if "granted=" in lineinfo[1]:
+                        granted = "granted=true" in lineinfo[1]
 
-            if current_user is not None:
-                user_property = re.match(
-                    r"(installReason|uninstallReason|dataDir|firstInstallTime)=(.*)",
-                    stripped,
-                )
-                if user_property:
-                    key = {
-                        "installReason": "install_reason",
-                        "uninstallReason": "uninstall_reason",
-                        "dataDir": "data_dir",
-                        "firstInstallTime": "first_install_time",
-                    }[user_property.group(1)]
-                    current_user[key] = _value(user_property.group(2))
-                    continue
+                    details["permissions"].append(
+                        {"name": permission, "granted": granted, "type": "install"}
+                    )
+            if in_runtime_permissions:
+                if not line.startswith(" " * 8):
+                    in_runtime_permissions = False
+                else:
+                    lineinfo = line.strip().split(":")
+                    permission = lineinfo[0]
+                    granted = None
+                    if "granted=" in lineinfo[1]:
+                        granted = "granted=true" in lineinfo[1]
 
-            if permission_section == "requested" and line.startswith("      "):
-                details["requested_permissions"].append(stripped)
-                continue
-            if permission_section in ("declared", "install") and line.startswith(
-                "      "
-            ):
-                details["permissions"].append(
-                    cls._permission(stripped, permission_section)
-                )
-                continue
-            if (
-                permission_section == "runtime"
-                and line.startswith("        ")
-                and current_user is not None
-            ):
-                current_user["permissions"].append(cls._permission(stripped, "runtime"))
-                continue
+                    runtime_permissions.setdefault(current_user, []).append(
+                        {"name": permission, "granted": granted, "type": "runtime"}
+                    )
+            if in_declared_permissions:
+                if not line.startswith(" " * 6):
+                    in_declared_permissions = False
+                else:
+                    permission = line.strip().split(":")[0]
+                    details["permissions"].append(
+                        {"name": permission, "type": "declared"}
+                    )
+            if in_requested_permissions:
+                if not line.startswith(" " * 6):
+                    in_requested_permissions = False
+                else:
+                    details["requested_permissions"].append(line.strip())
+            if line.strip().startswith("userId="):
+                details["uid"] = line.split("=")[1].strip()
+            elif line.strip().startswith("versionName="):
+                details["version_name"] = line.split("=")[1].strip()
+            elif line.strip().startswith("versionCode="):
+                details["version_code"] = line.split("=", 1)[1].strip()
+            elif line.strip().startswith("timeStamp="):
+                details["timestamp"] = line.split("=")[1].strip()
+            elif line.strip().startswith("installerPackageName="):
+                details["installer"] = line.split("=", 1)[1].strip()
+            elif line.strip().startswith("pkgFlags="):
+                details["system"] = "SYSTEM" in line.split("=", 1)[1].split()
+            elif line.strip().startswith("firstInstallTime="):
+                first_install_times[current_user] = line.split("=", 1)[1].strip()
+            elif line.strip().startswith("lastUpdateTime="):
+                details["last_update_time"] = line.split("=")[1].strip()
+            elif line.strip() == "install permissions:":
+                in_install_permissions = True
+            elif line.strip() == "runtime permissions:":
+                in_runtime_permissions = True
+            elif line.strip() == "declared permissions:":
+                in_declared_permissions = True
+            elif line.strip() == "requested permissions:":
+                in_requested_permissions = True
 
-            simple_match = re.match(
-                r"(appId|userId|versionName|timeStamp|lastUpdateTime|installerPackageName)=(.*)",
-                stripped,
-            )
-            if simple_match:
-                key = {
-                    "appId": "app_id",
-                    "userId": "app_id",
-                    "versionName": "version_name",
-                    "timeStamp": "timestamp",
-                    "lastUpdateTime": "last_update_time",
-                    "installerPackageName": "installer",
-                }[simple_match.group(1)]
-                raw_value = simple_match.group(2)
-                details[key] = (
-                    _value(raw_value)
-                    if key == "app_id" or raw_value == "null"
-                    else raw_value
-                )
-                continue
-            if stripped.startswith("pkgFlags="):
-                details["system"] = "SYSTEM" in stripped.split("=", 1)[1].split()
-                continue
-            version_match = re.match(
-                r"versionCode=([^\s]+)(?:\s+minSdk=([^\s]+))?(?:\s+targetSdk=([^\s]+))?",
-                stripped,
-            )
-            if version_match:
-                details["version_code"] = _value(version_match.group(1))
-                details["min_sdk"] = (
-                    _value(version_match.group(2)) if version_match.group(2) else None
-                )
-                details["target_sdk"] = (
-                    _value(version_match.group(3)) if version_match.group(3) else None
-                )
-            elif stripped.startswith("firstInstallTime="):
-                legacy_first_install = stripped.split("=", 1)[1]
+        if 0 in first_install_times:
+            details["first_install_time"] = first_install_times[0]
+        elif None in first_install_times:
+            details["first_install_time"] = first_install_times[None]
+        elif first_install_times:
+            details["first_install_time"] = next(iter(first_install_times.values()))
 
-        if legacy_first_install:
-            user_zero = next(
-                (user for user in details["users"] if user["user_id"] == 0), None
-            )
-            if user_zero is None:
-                user_zero = {"user_id": 0, "permissions": []}
-                details["users"].append(user_zero)
-            user_zero.setdefault("first_install_time", legacy_first_install)
+        if 0 in runtime_permissions:
+            details["permissions"].extend(runtime_permissions[0])
+        elif None in runtime_permissions:
+            details["permissions"].extend(runtime_permissions[None])
+        elif runtime_permissions:
+            details["permissions"].extend(next(iter(runtime_permissions.values())))
         return details
 
-    def parse(self, content: str) -> None:
-        self.results: list[dict[str, Any]] = []
-        category: str | None = None
-        package: dict[str, Any] | None = None
-        block: list[str] = []
+    def parse_dumpsys_packages(self, output: str) -> List[Dict[str, Any]]:
+        """
+        Parse the dumpsys package service data
+        """
+        pkg_rxp = re.compile(r"  Package \[(.+?)\].*")
 
-        def finish() -> None:
-            nonlocal package, block
-            if package is not None:
-                package.update(self.parse_dumpsys_package_for_details("\n".join(block)))
-                self.results.append(package)
-            package = None
-            block = []
+        results = []
+        package_name = None
+        package = {}
+        lines: list[str] = []
+        for line in output.splitlines():
+            if line.startswith("  Package ["):
+                if len(lines) > 0:
+                    details = self.parse_dumpsys_package_for_details("\n".join(lines))
+                    package.update(details)
+                    results.append(package)
+                    lines = []
+                    package = {}
 
+                matches = pkg_rxp.findall(line)
+                if not matches:
+                    continue
+
+                package_name = matches[0]
+                package["package_name"] = package_name
+                continue
+
+            if not package_name:
+                continue
+
+            lines.append(line)
+
+        if len(lines) > 0:
+            details = self.parse_dumpsys_package_for_details("\n".join(lines))
+            package.update(details)
+            results.append(package)
+
+        return results
+
+    def parse(self, content: str):
+        """
+        Parse the Dumpsys Package section for activities
+        Adds results to self.results
+
+        :param content: content of the package section (string)
+        """
+        self.results = []
+        package = []
+
+        in_package_list = False
         for line in content.splitlines():
-            if line == "Packages:":
-                finish()
-                category = "active"
+            if line.startswith("Packages:"):
+                in_package_list = True
                 continue
-            if line == "Hidden system packages:":
-                finish()
-                category = "hidden_system"
+
+            if not in_package_list:
                 continue
-            package_match = re.match(r"^  Package \[([^]]+)\]", line)
-            if package_match and category:
-                finish()
-                package = {
-                    "package_name": package_match.group(1),
-                    "package_type": category,
-                }
-                continue
-            if (
-                category
-                and line
-                and not line.startswith(" ")
-                and not line.endswith(" overlay paths:")
-            ):
-                finish()
-                category = None
-                continue
-            if package is not None:
-                block.append(line)
-        finish()
+
+            if line.strip() == "":
+                break
+
+            package.append(line)
+
+        self.results = self.parse_dumpsys_packages("\n".join(package))

src/mvt/android/artifacts/dumpsys_platform_compat.py

@@ -3,8 +3,6 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
-import re
-
 from .artifact import AndroidArtifact
 
 
@@ -27,37 +25,19 @@ class DumpsysPlatformCompatArtifact(AndroidArtifact):
 
     def parse(self, data: str) -> None:
         for line in data.splitlines():
-            match = re.match(r"ChangeId\((\d+);\s*(.*)\)$", line.strip())
-            if not match or "rawOverrides={" not in line:
+            if not line.startswith("ChangeId(168419799; name=DOWNSCALED;"):
                 continue
-            body = match.group(2)
-            name_match = re.search(r"(?:^|;\s*)name=([^;]+)", body)
-            state = (
-                "enabled"
-                if re.search(r"(?:^|;\s*)enabled(?:;|$)", body)
-                else "disabled"
-            )
-            overridable = bool(re.search(r"(?:^|;\s*)overridable(?:;|$)", body))
-            overrides_field = body.split("rawOverrides={", 1)[1].split("}", 1)[0]
-            for entry in overrides_field.split(","):
-                package_name, separator, raw_value = entry.strip().partition("=")
-                if not separator:
-                    continue
-                value: bool | int | str
-                if raw_value in ("true", "false"):
-                    value = raw_value == "true"
-                else:
-                    try:
-                        value = int(raw_value)
-                    except ValueError:
-                        value = raw_value
-                self.results.append(
-                    {
-                        "change_id": int(match.group(1)),
-                        "change_name": name_match.group(1) if name_match else None,
-                        "change_state": state,
-                        "overridable": overridable,
-                        "package_name": package_name,
-                        "override_value": value,
-                    }
-                )
+
+            if line.strip() == "":
+                break
+
+            # Look for rawOverrides field
+            if "rawOverrides={" in line:
+                # Extract the content inside the braces for rawOverrides
+                overrides_field = line.split("rawOverrides={", 1)[1].split("};", 1)[0]
+
+                for entry in overrides_field.split(", "):
+                    # Extract app name
+                    uninstall_app = entry.split("=")[0].strip()
+
+                    self.results.append({"package_name": uninstall_app})

src/mvt/android/artifacts/dumpsys_receivers.py

@@ -4,7 +4,6 @@
 #   https://license.mvt.re/1.1/
 
 from .artifact import AndroidArtifact
-from .package_resolvers import parse_resolver_table
 
 INTENT_NEW_OUTGOING_SMS = "android.provider.Telephony.NEW_OUTGOING_SMS"
 INTENT_SMS_RECEIVED = "android.provider.Telephony.SMS_RECEIVED"
@@ -19,45 +18,115 @@ class DumpsysReceiversArtifact(AndroidArtifact):
     """
 
     def check_indicators(self) -> None:
-        for receiver in self.results:
-            intent = receiver["key"]
-            if intent == INTENT_NEW_OUTGOING_SMS:
-                self.log.info(
-                    'Found a receiver to intercept outgoing SMS messages: "%s"',
-                    receiver["component"],
-                )
-            elif intent == INTENT_SMS_RECEIVED:
-                self.log.info(
-                    'Found a receiver to intercept incoming SMS messages: "%s"',
-                    receiver["component"],
-                )
-            elif intent == INTENT_DATA_SMS_RECEIVED:
-                self.log.info(
-                    'Found a receiver to intercept incoming data SMS message: "%s"',
-                    receiver["component"],
-                )
-            elif intent == INTENT_PHONE_STATE:
-                self.log.info(
-                    'Found a receiver monitoring telephony state/incoming calls: "%s"',
-                    receiver["component"],
-                )
-            elif intent == INTENT_NEW_OUTGOING_CALL:
-                self.log.info(
-                    'Found a receiver monitoring outgoing calls: "%s"',
-                    receiver["component"],
-                )
+        for intent, receivers in self.results.items():
+            for receiver in receivers:
+                if intent == INTENT_NEW_OUTGOING_SMS:
+                    self.log.info(
+                        'Found a receiver to intercept outgoing SMS messages: "%s"',
+                        receiver["receiver"],
+                    )
+                elif intent == INTENT_SMS_RECEIVED:
+                    self.log.info(
+                        'Found a receiver to intercept incoming SMS messages: "%s"',
+                        receiver["receiver"],
+                    )
+                elif intent == INTENT_DATA_SMS_RECEIVED:
+                    self.log.info(
+                        'Found a receiver to intercept incoming data SMS message: "%s"',
+                        receiver["receiver"],
+                    )
+                elif intent == INTENT_PHONE_STATE:
+                    self.log.info(
+                        "Found a receiver monitoring "
+                        'telephony state/incoming calls: "%s"',
+                        receiver["receiver"],
+                    )
+                elif intent == INTENT_NEW_OUTGOING_CALL:
+                    self.log.info(
+                        'Found a receiver monitoring outgoing calls: "%s"',
+                        receiver["receiver"],
+                    )
 
-            if not self.indicators:
-                continue
+                if not self.indicators:
+                    continue
 
-            ioc_match = self.indicators.check_app_id(receiver["package_name"])
-            if ioc_match:
-                self.alertstore.critical(
-                    ioc_match.message,
-                    "",
-                    receiver,
-                    matched_indicator=ioc_match.ioc,
-                )
+                ioc_match = self.indicators.check_app_id(receiver["package_name"])
+                if ioc_match:
+                    self.alertstore.critical(
+                        ioc_match.message,
+                        "",
+                        {intent: receiver},
+                        matched_indicator=ioc_match.ioc,
+                    )
+                    continue
 
     def parse(self, output: str) -> None:
-        self.results = parse_resolver_table(output, "Receiver")
+        self.results: dict[str, list[dict[str, str]]] = {}
+
+        in_receiver_resolver_table = False
+        in_non_data_actions = False
+        intent = None
+        for line in output.splitlines():
+            if line.startswith("Receiver Resolver Table:"):
+                in_receiver_resolver_table = True
+                continue
+
+            if not in_receiver_resolver_table:
+                continue
+
+            if line.startswith("  Non-Data Actions:"):
+                in_non_data_actions = True
+                continue
+
+            if not in_non_data_actions:
+                continue
+
+            # If we hit an empty line, the Non-Data Actions section should be
+            # finished.
+            if line.strip() == "":
+                break
+
+            # We detect the action name.
+            if (
+                line.startswith(" " * 6)
+                and not line.startswith(" " * 8)
+                and ":" in line
+            ):
+                intent = line.strip().replace(":", "")
+                self.results[intent] = []
+                continue
+
+            parts = line.strip().split(" ")
+            if len(parts) < 2:
+                # A single-token line here is not a receiver. Real dumpstate
+                # output can print an action header mis-indented (observed with
+                # 15 leading spaces instead of 6), which used to raise
+                # IndexError and abort the whole module. Treat a trailing-colon
+                # token as the next action, skip anything else.
+                if parts[0].endswith(":"):
+                    intent = parts[0][:-1]
+                    self.results.setdefault(intent, [])
+                continue
+
+            # If we are not in an intent block yet, skip.
+            if not intent:
+                continue
+
+            # If we are in a block but the line does not start with 8 spaces
+            # it means the block ended a new one started, so we reset and
+            # continue.
+            if not line.startswith(" " * 8):
+                intent = None
+                continue
+
+            # If we got this far, we are processing receivers for the
+            # activities we are interested in.
+            receiver = parts[1]
+            package_name = receiver.split("/")[0]
+
+            self.results[intent].append(
+                {
+                    "package_name": package_name,
+                    "receiver": receiver,
+                }
+            )

src/mvt/android/artifacts/getprop.py

@@ -28,18 +28,18 @@ INTERESTING_PROPERTIES = [
 class GetProp(AndroidArtifact):
     def parse(self, entry: str) -> None:
         self.results: List[Dict[str, str]] = []
-        rxp = re.compile(r"^\[([^]]+)\]: \[(.*)\]$")
+        rxp = re.compile(r"\[(.+?)\]: \[(.+?)\]")
 
         for line in entry.splitlines():
             line = line.strip()
             if line == "":
                 continue
 
-            match = rxp.match(line)
-            if not match:
+            matches = re.findall(rxp, line)
+            if not matches or len(matches[0]) != 2:
                 continue
 
-            prop_entry = {"name": match.group(1), "value": match.group(2)}
+            prop_entry = {"name": matches[0][0], "value": matches[0][1]}
             self.results.append(prop_entry)
 
     def get_device_timezone(self) -> str | None:

src/mvt/android/artifacts/mounts.py

@@ -117,51 +117,6 @@ class Mounts(AndroidArtifact):
                 # Skip lines that don't match expected format
                 continue
 
-    @staticmethod
-    def parse_mountinfo(entry: str, process_id: int) -> list[dict[str, Any]]:
-        """Parse Linux /proc/PID/mountinfo records."""
-        results = []
-        for line in entry.splitlines():
-            fields = line.split()
-            if "-" not in fields:
-                continue
-            separator = fields.index("-")
-            if separator < 6 or len(fields) < separator + 4:
-                continue
-            try:
-                mount_id = int(fields[0])
-                parent_id = int(fields[1])
-            except ValueError:
-                continue
-            mount_options = fields[5].split(",")
-            super_options = fields[separator + 3].split(",")
-            options = list(dict.fromkeys(mount_options + super_options))
-            mount_point = fields[4].replace("\\040", " ")
-            device = fields[separator + 2].replace("\\040", " ")
-            filesystem_type = fields[separator + 1]
-            is_system = mount_point in SUSPICIOUS_MOUNT_POINTS or any(
-                mount_point.startswith(f"{prefix}/")
-                for prefix in SUSPICIOUS_MOUNT_POINTS
-            )
-            results.append(
-                {
-                    "mount_id": mount_id,
-                    "parent_id": parent_id,
-                    "major_minor": fields[2],
-                    "root": fields[3].replace("\\040", " "),
-                    "mount_point": mount_point,
-                    "device": device,
-                    "filesystem_type": filesystem_type,
-                    "mount_options": ",".join(options),
-                    "options_list": options,
-                    "optional_fields": fields[6:separator],
-                    "is_system_partition": is_system,
-                    "is_read_write": "rw" in options,
-                    "process_ids": [process_id],
-                }
-            )
-        return results
-
     def check_indicators(self) -> None:
         """
         Check for suspicious mount configurations that may indicate root access

src/mvt/android/artifacts/package_resolvers.py

@@ -1,70 +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 re
-
-
-RESOLVER_TYPES = {
-    "Full MIME Types": "full_mime_type",
-    "Base MIME Types": "base_mime_type",
-    "Wild MIME Types": "wild_mime_type",
-    "Schemes": "scheme",
-    "Non-Data Actions": "non_data_action",
-    "MIME Typed Actions": "mime_typed_action",
-}
-
-ENTRY_RE = re.compile(
-    r"^\s{8,}[0-9a-fA-F]+\s+(?P<component>\S+)"
-    r"(?:\s+\((?P<filters>\d+)\s+filters?\))?\s*$"
-)
-
-
-def parse_resolver_table(content: str, table_name: str) -> list[dict]:
-    """Parse every resolver category from a dumpsys package table."""
-    results: list[dict] = []
-    in_table = False
-    resolver_type: str | None = None
-    key: str | None = None
-
-    for line in content.splitlines():
-        if line.startswith(f"{table_name} Resolver Table:"):
-            in_table = True
-            continue
-        if not in_table:
-            continue
-
-        if line and not line.startswith(" "):
-            break
-
-        section_match = re.match(r"^ {2}([^ ].*):\s*$", line)
-        if section_match:
-            resolver_type = RESOLVER_TYPES.get(section_match.group(1))
-            key = None
-            continue
-        if resolver_type is None:
-            continue
-
-        key_match = re.match(r"^ {6,}([^ ].*):\s*$", line)
-        if key_match:
-            key = key_match.group(1)
-            continue
-        if key is None:
-            continue
-
-        entry_match = ENTRY_RE.match(line)
-        if not entry_match:
-            continue
-        component = entry_match.group("component")
-        results.append(
-            {
-                "resolver_type": resolver_type,
-                "key": key,
-                "package_name": component.split("/", 1)[0],
-                "component": component,
-                "filter_count": int(entry_match.group("filters") or 1),
-            }
-        )
-
-    return results

src/mvt/android/artifacts/processes.py

@@ -1,91 +1,72 @@
 # Mobile Verification Toolkit (MVT)
-# Copyright (c) 2021-2026 The MVT Authors.
+# 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/
 
 from .artifact import AndroidArtifact
 
 
-FIELD_NAMES = {
-    "LABEL": "label",
-    "USER": "user",
-    "PID": "pid",
-    "TID": "tid",
-    "PPID": "ppid",
-    "VSZ": "virtual_memory_size",
-    "RSS": "resident_set_size",
-    "WCHAN": "wchan",
-    "ADDR": "address",
-    "S": "state",
-    "PRI": "priority",
-    "NI": "nice",
-    "RTPRIO": "realtime_priority",
-    "SCH": "scheduler",
-    "PCY": "policy",
-    "TIME": "cpu_time",
-    "CMD": "command",
-    "NAME": "command",
-}
-INTEGER_FIELDS = {
-    "pid",
-    "tid",
-    "ppid",
-    "virtual_memory_size",
-    "resident_set_size",
-    "priority",
-    "nice",
-}
-
-
 class Processes(AndroidArtifact):
     def parse(self, entry: str) -> None:
-        self.results = []
-        lines = [line for line in entry.splitlines() if line.strip()]
-        if not lines:
-            return
-        headers = lines[0].split()
-        if not all(header in FIELD_NAMES for header in headers):
-            return
+        for line in entry.splitlines()[1:]:
+            proc = line.split()
 
-        for line in lines[1:]:
-            values = line.split(None, len(headers) - 1)
-            if len(values) != len(headers):
+            # Skip empty lines
+            if len(proc) == 0:
                 continue
-            result = {}
-            valid = True
-            for header, raw in zip(headers, values):
-                key = FIELD_NAMES[header]
-                value: str | int = raw.strip("[]") if key == "command" else raw
-                if key in INTEGER_FIELDS:
-                    try:
-                        value = int(value)
-                    except ValueError:
-                        valid = False
-                        break
-                result[key] = value
-            if valid:
-                self.results.append(result)
+
+            # Sometimes WCHAN is empty.
+            if len(proc) == 8:
+                proc = proc[:5] + [""] + proc[5:]
+
+            # Sometimes there is the security label.
+            if proc[0].startswith("u:r"):
+                label = proc[0]
+                proc = proc[1:]
+            else:
+                label = ""
+
+            # Sometimes there is no WCHAN.
+            if len(proc) < 9:
+                proc = proc[:5] + [""] + proc[5:]
+
+            self.results.append(
+                {
+                    "user": proc[0],
+                    "pid": int(proc[1]),
+                    "ppid": int(proc[2]),
+                    "virtual_memory_size": int(proc[3]),
+                    "resident_set_size": int(proc[4]),
+                    "wchan": proc[5],
+                    "aprocress": proc[6],
+                    "stat": proc[7],
+                    "proc_name": proc[8].strip("[]"),
+                    "label": label,
+                }
+            )
 
     def check_indicators(self) -> None:
         if not self.indicators:
             return
+
         for result in self.results:
-            command = result.get("command", "")
-            if not isinstance(command, str):
+            proc_name = result.get("proc_name", "")
+            if not proc_name:
                 continue
-            process_name = command.rsplit("/", 1)[-1]
-            if not process_name or process_name == "gatekeeperd":
+
+            # Skipping this process because of false positives.
+            if result["proc_name"] == "gatekeeperd":
                 continue
-            for checker in (
-                self.indicators.check_app_id,
-                self.indicators.check_process,
-            ):
-                ioc_match = checker(process_name)
-                if ioc_match:
-                    self.alertstore.critical(
-                        ioc_match.message,
-                        "",
-                        result,
-                        matched_indicator=ioc_match.ioc,
-                    )
-                    break
+
+            ioc_match = self.indicators.check_app_id(proc_name)
+            if ioc_match:
+                self.alertstore.critical(
+                    ioc_match.message, "", result, matched_indicator=ioc_match.ioc
+                )
+                continue
+
+            ioc_match = self.indicators.check_process(proc_name)
+            if ioc_match:
+                self.alertstore.critical(
+                    ioc_match.message, "", result, matched_indicator=ioc_match.ioc
+                )

src/mvt/android/artifacts/settings.py

@@ -3,8 +3,6 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
-import re
-
 from .artifact import AndroidArtifact
 
 ANDROID_DANGEROUS_SETTINGS = [
@@ -62,28 +60,6 @@ ANDROID_DANGEROUS_SETTINGS = [
 
 
 class Settings(AndroidArtifact):
-    def parse(self, content: str) -> None:
-        self.results: dict[str, dict[str, str]] = {}
-        namespace: str | None = None
-        for line in content.splitlines():
-            heading = re.match(
-                r"^(CONFIG|GLOBAL|SECURE|SYSTEM) SETTINGS \(user (\d+)\)$",
-                line.strip(),
-            )
-            if heading:
-                namespace = f"{heading.group(1).lower()}:user_{heading.group(2)}"
-                self.results[namespace] = {}
-                continue
-            if namespace is None or not line.startswith("_id:"):
-                continue
-            setting = re.match(
-                r"^_id:\S+\s+name:(.*?)\s+pkg:.*?\s+value:(.*?)"
-                r"(?:\s+default:.*\s+defaultSystemSet:(?:true|false))?$",
-                line,
-            )
-            if setting:
-                self.results[namespace][setting.group(1)] = setting.group(2)
-
     def check_indicators(self) -> None:
         for namespace, settings in self.results.items():
             for key, value in settings.items():

src/mvt/android/artifacts/tombstone_crashes.py

@@ -131,13 +131,6 @@ class TombstoneCrashArtifact(AndroidArtifact):
         self, file_name: str, file_timestamp: datetime.datetime, data: bytes
     ) -> None:
         """Parse Android tombstone crash files from a protobuf object."""
-        self.results.append(self.parse_protobuf_record(file_name, file_timestamp, data))
-
-    def parse_protobuf_record(
-        self, file_name: str, file_timestamp: datetime.datetime, data: bytes
-    ) -> dict:
-        if not data:
-            raise ValueError("empty protobuf tombstone")
         tombstone_pb = Tombstone().parse(data)
         tombstone_dict = tombstone_pb.to_dict(
             casing=betterproto2.Casing.SNAKE, include_default_values=True
@@ -150,31 +143,20 @@ class TombstoneCrashArtifact(AndroidArtifact):
         tombstone_dict["file_name"] = file_name
         tombstone_dict["file_timestamp"] = convert_datetime_to_iso(file_timestamp)
         tombstone_dict["process_name"] = self._proccess_name_from_thread(tombstone_dict)
-        if isinstance(tombstone_dict.get("selinux_label"), str):
-            tombstone_dict["selinux_label"] = tombstone_dict["selinux_label"].rstrip(
-                "\x00"
-            )
 
         # Confirm the tombstone is valid, and matches the output model
         tombstone = TombstoneCrashResult.model_validate(tombstone_dict)
-        return tombstone.model_dump()
+        self.results.append(tombstone.model_dump())
 
     def parse(
         self, file_name: str, file_timestamp: datetime.datetime, content: bytes
     ) -> None:
         """Parse text Android tombstone crash files."""
-        self.results.append(self.parse_text_record(file_name, file_timestamp, content))
-
-    def parse_text_record(
-        self, file_name: str, file_timestamp: datetime.datetime, content: bytes
-    ) -> dict:
-        if not content:
-            raise ValueError("empty plaintext tombstone")
         tombstone_dict = {
             "file_name": file_name,
             "file_timestamp": convert_datetime_to_iso(file_timestamp),
         }
-        lines = content.decode("utf-8", errors="replace").splitlines()
+        lines = content.decode("utf-8").splitlines()
         for line_num, line in enumerate(lines, 1):
             if not line.strip() or TOMBSTONE_DELIMITER in line:
                 continue
@@ -189,7 +171,7 @@ class TombstoneCrashArtifact(AndroidArtifact):
 
         # Validate the tombstone and add it to the results
         tombstone = TombstoneCrashResult.model_validate(tombstone_dict)
-        return tombstone.model_dump()
+        self.results.append(tombstone.model_dump())
 
     def _parse_tombstone_line(
         self, line: str, key: str, destination_key: str, tombstone: dict
@@ -213,9 +195,7 @@ class TombstoneCrashArtifact(AndroidArtifact):
         if line_key != key:
             raise ValueError(f"Expected key {key}, got {line_key}")
 
-        value_clean = value.strip()
-        if len(value_clean) >= 2 and value_clean[0] == value_clean[-1] == "'":
-            value_clean = value_clean[1:-1]
+        value_clean = value.strip().strip("'")
         if destination_key == "uid":
             tombstone[destination_key] = int(value_clean)
         elif destination_key == "process_uptime":
@@ -289,7 +269,9 @@ class TombstoneCrashArtifact(AndroidArtifact):
     @staticmethod
     def _parse_timestamp_string(timestamp: str) -> str:
         timestamp_parsed = parser.parse(timestamp)
-        return convert_datetime_to_iso(timestamp_parsed)
+        # Preserve the source wall-clock time while returning the project-wide ISO format.
+        local_timestamp = timestamp_parsed.replace(tzinfo=datetime.timezone.utc)
+        return convert_datetime_to_iso(local_timestamp)
 
     @staticmethod
     def _proccess_name_from_thread(tombstone_dict: dict) -> str:

src/mvt/android/cmd_check_bugreport.py

@@ -53,7 +53,6 @@ class CmdAndroidCheckBugreport(Command):
 
         self.platform = "android"
         self.name = "check-bugreport"
-        self.output_schema_version = 2
         self.modules = BUGREPORT_MODULES
 
         self.__format: str = ""

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

@@ -3,8 +3,6 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
-from mvt.common.module import MVTModule
-
 from .aqf_files import AQFFiles
 from .aqf_getprop import AQFGetProp
 from .aqf_log_timestamps import AQFLogTimestamps
@@ -14,7 +12,7 @@ from .aqf_settings import AQFSettings
 from .mounts import Mounts
 from .root_binaries import RootBinaries
 
-ANDROIDQF_MODULES: list[type[MVTModule]] = [
+ANDROIDQF_MODULES = [
     AQFPackages,
     AQFProcesses,
     AQFGetProp,

src/mvt/android/modules/backup/__init__.py

@@ -3,8 +3,6 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
-from mvt.common.module import MVTModule
-
 from .sms import SMS
 
-BACKUP_MODULES: list[type[MVTModule]] = [SMS]
+BACKUP_MODULES = [SMS]

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

@@ -3,8 +3,6 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
-from mvt.common.module import MVTModule
-
 from .dumpsys_accessibility import DumpsysAccessibility
 from .dumpsys_activities import DumpsysActivities
 from .dumpsys_appops import DumpsysAppops
@@ -18,11 +16,8 @@ from .dumpsys_receivers import DumpsysReceivers
 from .dumpsys_adb_state import DumpsysADBState
 from .fs_timestamps import BugReportTimestamps
 from .tombstones import Tombstones
-from .mounts import Mounts
-from .processes import Processes
-from .settings import Settings
 
-BUGREPORT_MODULES: list[type[MVTModule]] = [
+BUGREPORT_MODULES = [
     DumpsysAccessibility,
     DumpsysActivities,
     DumpsysAppops,
@@ -36,7 +31,4 @@ BUGREPORT_MODULES: list[type[MVTModule]] = [
     DumpsysADBState,
     BugReportTimestamps,
     Tombstones,
-    Processes,
-    Settings,
-    Mounts,
 ]

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

@@ -72,11 +72,7 @@ class BugReportModule(MVTModule):
             if not self.extract_path:
                 raise ValueError("extract_path is not set")
             joined = os.path.join(self.extract_path, file_path)
-            if (
-                not Path(joined)
-                .resolve()
-                .is_relative_to(Path(self.extract_path).resolve())
-            ):
+            if not Path(joined).resolve().is_relative_to(Path(self.extract_path).resolve()):
                 raise ValueError("unsafe file_path")
             handle = open(joined, "rb")
 
@@ -104,27 +100,6 @@ class BugReportModule(MVTModule):
 
         return None
 
-    @staticmethod
-    def extract_command_section(content: str, heading: str) -> str:
-        """Return a bugreport command section without consuming the next one.
-
-        Bugreport separators include timing text, so looking for a line equal to
-        ``------`` is not sufficient and can accidentally feed the remainder of
-        dumpstate to a parser.
-        """
-        lines: list[str] = []
-        in_section = False
-        for line in content.splitlines():
-            stripped = line.strip()
-            if not in_section:
-                if stripped.startswith(heading):
-                    in_section = True
-                continue
-            if stripped.startswith("------"):
-                break
-            lines.append(line)
-        return "\n".join(lines)
-
     def _get_file_modification_time(self, file_path: str) -> datetime.datetime:
         if self.zip_archive:
             file_timetuple = self.zip_archive.getinfo(file_path).date_time

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

@@ -49,7 +49,9 @@ class DumpsysAccessibility(DumpsysAccessibilityArtifact, BugReportModule):
         self.parse(content)
 
         for result in self.results:
-            self.log.info('Found accessibility service "%s"', result.get("component"))
+            self.log.info(
+                'Found installed accessibility service "%s"', result.get("service")
+            )
 
         self.log.info(
             "Identified a total of %d accessibility services", len(self.results)

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

@@ -44,8 +44,21 @@ class DumpsysGetProp(GetPropArtifact, BugReportModule):
             )
             return
 
-        section = self.extract_command_section(
-            content.decode(errors="ignore"), "------ SYSTEM PROPERTIES"
-        )
-        self.parse(section)
+        lines = []
+        in_getprop = False
+
+        for line in content.decode(errors="ignore").splitlines():
+            if line.strip().startswith("------ SYSTEM PROPERTIES"):
+                in_getprop = True
+                continue
+
+            if not in_getprop:
+                continue
+
+            if line.strip() == "------":
+                break
+
+            lines.append(line)
+
+        self.parse("\n".join(lines))
         self.log.info("Extracted %d Android system properties", len(self.results))

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

@@ -50,10 +50,7 @@ class DumpsysPackages(DumpsysPackagesArtifact, BugReportModule):
 
         for result in self.results:
             dangerous_permissions_count = 0
-            permissions = list(result["permissions"])
-            for user in result.get("users", []):
-                permissions.extend(user.get("permissions", []))
-            for perm in permissions:
+            for perm in result["permissions"]:
                 if perm["name"] in DANGEROUS_PERMISSIONS:
                     dangerous_permissions_count += 1
 

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

@@ -13,7 +13,7 @@ from mvt.common.module_types import ModuleResults
 
 
 class DumpsysPlatformCompat(DumpsysPlatformCompatArtifact, BugReportModule):
-    """This module extracts raw per-package compatibility overrides."""
+    """This module extracts details on uninstalled apps."""
 
     def __init__(
         self,
@@ -48,4 +48,4 @@ class DumpsysPlatformCompat(DumpsysPlatformCompatArtifact, BugReportModule):
         )
         self.parse(content)
 
-        self.log.info("Found %d package compatibility overrides", len(self.results))
+        self.log.info("Found %d uninstalled apps", len(self.results))

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

@@ -33,7 +33,7 @@ class DumpsysReceivers(DumpsysReceiversArtifact, BugReportModule):
             results=results,
         )
 
-        self.results = results if results else []
+        self.results = results if results else {}
 
     def run(self) -> None:
         content = self._get_dumpstate_file()
@@ -49,4 +49,4 @@ class DumpsysReceivers(DumpsysReceiversArtifact, BugReportModule):
         )
 
         self.parse(dumpsys_section)
-        self.log.info("Extracted %d package receivers", len(self.results))
+        self.log.info("Extracted receivers for %d intents", len(self.results))

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

@@ -4,15 +4,12 @@
 #   https://license.mvt.re/1.1/
 
 import logging
-import datetime
 from typing import Optional
-from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 
 from mvt.common.utils import convert_datetime_to_iso
 from .base import BugReportModule
 from mvt.common.module_types import ModuleResults
 from mvt.android.artifacts.file_timestamps import FileTimestampsArtifact
-from mvt.android.artifacts.getprop import GetProp
 
 
 class BugReportTimestamps(FileTimestampsArtifact, BugReportModule):
@@ -41,44 +38,15 @@ class BugReportTimestamps(FileTimestampsArtifact, BugReportModule):
     def run(self) -> None:
         filesystem_files = self._get_files_by_pattern("FS/*")
 
-        timezone_name = None
-        dumpstate = self._get_dumpstate_file()
-        if dumpstate:
-            section = self.extract_command_section(
-                dumpstate.decode("utf-8", errors="replace"),
-                "------ SYSTEM PROPERTIES",
-            )
-            properties = GetProp()
-            properties.parse(section)
-            timezone_name = properties.get_device_timezone()
-        timezone = None
-        if timezone_name:
-            try:
-                timezone = ZoneInfo(timezone_name)
-            except ZoneInfoNotFoundError:
-                self.log.warning("Unknown device timezone %s", timezone_name)
-
         self.results = []
         for file in filesystem_files:
             # Only the modification time is available in the zip file metadata.
             # The timezone is the local timezone of the machine the phone.
             modification_time = self._get_file_modification_time(file)
-            utc_time = None
-            if timezone is not None:
-                utc_time = convert_datetime_to_iso(
-                    modification_time.replace(tzinfo=timezone).astimezone(
-                        datetime.timezone.utc
-                    )
-                )
             self.results.append(
                 {
                     "path": file,
                     "modified_time": convert_datetime_to_iso(modification_time),
-                    "modified_time_utc": utc_time,
-                    "timezone": timezone_name,
-                    "timestamp_source": (
-                        "zip_metadata" if self.zip_archive else "filesystem_metadata"
-                    ),
                 }
             )
 

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

@@ -1,41 +0,0 @@
-# Mobile Verification Toolkit (MVT)
-# Copyright (c) 2021-2026 The MVT Authors.
-
-import re
-
-from mvt.android.artifacts.mounts import Mounts as MountsArtifact
-
-from .base import BugReportModule
-
-
-class Mounts(MountsArtifact, BugReportModule):
-    """Extract and deduplicate process mount namespaces from mountinfo."""
-
-    def run(self) -> None:
-        mount_files = self._get_files_by_pattern("FS/proc/*/mountinfo")
-        if not mount_files:
-            mount_files = self._get_files_by_pattern("*/proc/*/mountinfo")
-        unique: dict[tuple, dict] = {}
-        for file_path in mount_files:
-            pid_match = re.search(r"/proc/(\d+)/mountinfo$", file_path)
-            if not pid_match:
-                continue
-            entries = self.parse_mountinfo(
-                self._get_file_content(file_path).decode("utf-8", errors="replace"),
-                int(pid_match.group(1)),
-            )
-            for entry in entries:
-                identity = (
-                    entry["major_minor"],
-                    entry["root"],
-                    entry["mount_point"],
-                    entry["device"],
-                    entry["filesystem_type"],
-                    entry["mount_options"],
-                )
-                if identity in unique:
-                    unique[identity]["process_ids"].extend(entry["process_ids"])
-                else:
-                    unique[identity] = entry
-        self.results = list(unique.values())
-        self.log.info("Extracted %d unique mount records", len(self.results))

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

@@ -1,22 +0,0 @@
-# Mobile Verification Toolkit (MVT)
-# Copyright (c) 2021-2026 The MVT Authors.
-
-from mvt.android.artifacts.processes import Processes as ProcessesArtifact
-
-from .base import BugReportModule
-
-
-class Processes(ProcessesArtifact, BugReportModule):
-    """Extract the process and thread table from dumpstate."""
-
-    def run(self) -> None:
-        data = self._get_dumpstate_file()
-        if not data:
-            self.log.error("Unable to find dumpstate file")
-            return
-        section = self.extract_command_section(
-            data.decode("utf-8", errors="replace"),
-            "------ PROCESSES AND THREADS",
-        )
-        self.parse(section)
-        self.log.info("Identified %d running process threads", len(self.results))

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

@@ -1,22 +0,0 @@
-# Mobile Verification Toolkit (MVT)
-# Copyright (c) 2021-2026 The MVT Authors.
-
-from mvt.android.artifacts.settings import Settings as SettingsArtifact
-
-from .base import BugReportModule
-
-
-class Settings(SettingsArtifact, BugReportModule):
-    """Extract all SettingsProvider namespaces and users."""
-
-    def run(self) -> None:
-        data = self._get_dumpstate_file()
-        if not data:
-            self.log.error("Unable to find dumpstate file")
-            return
-        section = self.extract_dumpsys_section(
-            data.decode("utf-8", errors="replace"), "DUMP OF SERVICE settings:"
-        )
-        self.parse(section)
-        count = sum(len(settings) for settings in self.results.values())
-        self.log.info("Identified %d Android settings", count)

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

@@ -43,76 +43,21 @@ class Tombstones(TombstoneCrashArtifact, BugReportModule):
             )
             return
 
-        grouped: dict[str, dict[str, str]] = {}
-        for file_path in tombstone_files:
-            file_name = file_path.rsplit("/", 1)[-1]
-            source = "protobuf" if file_name.endswith(".pb") else "text"
-            crash_id = file_name.removesuffix(".pb")
-            grouped.setdefault(crash_id, {})[source] = file_path
+        for tombstone_file in sorted(tombstone_files):
+            tombstone_filename = tombstone_file.split("/")[-1]
+            modification_time = self._get_file_modification_time(tombstone_file)
+            tombstone_data = self._get_file_content(tombstone_file)
 
-        for crash_id, paths in sorted(grouped.items()):
-            parsed_sources: dict[str, dict] = {}
-            source_records: dict[str, dict] = {}
-            for source in ("text", "protobuf"):
-                file_path = paths.get(source)
-                if file_path is None:
-                    continue
-                file_name = file_path.rsplit("/", 1)[-1]
-                file_timestamp = self._get_file_modification_time(file_path)
-                source_info = {
-                    "file_name": file_name,
-                    "file_timestamp": file_timestamp.isoformat(),
-                    "parsed": False,
-                    "error": None,
-                    "record": None,
-                }
-                try:
-                    data = self._get_file_content(file_path)
-                    if source == "protobuf":
-                        record = self.parse_protobuf_record(
-                            file_name, file_timestamp, data
-                        )
-                    else:
-                        record = self.parse_text_record(file_name, file_timestamp, data)
-                    source_info["parsed"] = True
-                    source_info["record"] = record
-                    source_records[source] = record
-                except Exception as exc:
-                    source_info["error"] = str(exc)
-                    self.log.error(
-                        "Error parsing tombstone file %s: %s", file_path, exc
+            try:
+                if tombstone_file.endswith(".pb"):
+                    self.parse_protobuf(
+                        tombstone_filename, modification_time, tombstone_data
                     )
-                parsed_sources[source] = source_info
-
-            if not source_records:
-                continue
-            preferred = source_records.get("protobuf") or source_records["text"]
-            canonical = dict(preferred)
-            text_record = source_records.get("text")
-            if text_record:
-                for key, value in text_record.items():
-                    if canonical.get(key) in (None, "", [], {}):
-                        canonical[key] = value
-
-            differences = {}
-            protobuf_record = source_records.get("protobuf")
-            if text_record and protobuf_record:
-                for key in text_record.keys() & protobuf_record.keys():
-                    if key in ("file_name", "file_timestamp"):
-                        continue
-                    if text_record[key] != protobuf_record[key]:
-                        differences[key] = {
-                            "text": text_record[key],
-                            "protobuf": protobuf_record[key],
-                        }
-            canonical.update(
-                {
-                    "crash_id": crash_id,
-                    "sources": parsed_sources,
-                    "differences": differences,
-                }
-            )
-            self.results.append(canonical)
+                else:
+                    self.parse(tombstone_filename, modification_time, tombstone_data)
+            except ValueError as e:
+                # Catch any exceptions raised during parsing or validation.
+                self.log.error(f"Error parsing tombstone file {tombstone_file}: {e}")
 
         self.log.info(
             "Extracted a total of %d tombstone files",

src/mvt/android/modules/intrusion_logs/__init__.py

@@ -3,13 +3,11 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
-from mvt.common.module import MVTModule
-
 from .connect_event import ConnectEvent
 from .dns_event import DnsEvent
 from .security_event import SecurityEvent
 
-INTRUSION_LOGS_MODULES: list[type[MVTModule]] = [
+INTRUSION_LOGS_MODULES = [
     DnsEvent,
     ConnectEvent,
     SecurityEvent,

src/mvt/common/command.py

@@ -54,7 +54,6 @@ class Command:
     ) -> None:
         self.name = ""
         self.platform = ""
-        self.output_schema_version: Optional[int] = None
         self.modules: list[type[MVTModule]] = []
         self.custom_modules = custom_modules if custom_modules else []
 
@@ -187,8 +186,6 @@ class Command:
             "ioc_files": [],
             "hashes": [],
         }
-        if self.output_schema_version is not None:
-            info["output_schema_version"] = self.output_schema_version
 
         for coll in self.iocs.ioc_collections:
             ioc_file_path = coll.get("stix2_file_path", "")

src/mvt/ios/modules/backup/__init__.py

@@ -3,16 +3,9 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
-from mvt.common.module import MVTModule
-
 from .backup_info import BackupInfo
 from .configuration_profiles import ConfigurationProfiles
 from .manifest import Manifest
 from .profile_events import ProfileEvents
 
-BACKUP_MODULES: list[type[MVTModule]] = [
-    BackupInfo,
-    ConfigurationProfiles,
-    Manifest,
-    ProfileEvents,
-]
+BACKUP_MODULES = [BackupInfo, ConfigurationProfiles, Manifest, ProfileEvents]

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

@@ -3,8 +3,6 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
-from mvt.common.module import MVTModule
-
 from .analytics import Analytics
 from .analytics_ios_versions import AnalyticsIOSVersions
 from .cache_files import CacheFiles
@@ -17,7 +15,7 @@ from .webkit_indexeddb import WebkitIndexedDB
 from .webkit_localstorage import WebkitLocalStorage
 from .webkit_safariviewservice import WebkitSafariViewService
 
-FS_MODULES: list[type[MVTModule]] = [
+FS_MODULES = [
     CacheFiles,
     Filesystem,
     Netusage,

src/mvt/ios/modules/mixed/__init__.py

@@ -3,8 +3,6 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
-from mvt.common.module import MVTModule
-
 from .applications import Applications
 from .calendar import Calendar
 from .calls import Calls
@@ -30,7 +28,7 @@ from .webkit_session_resource_log import WebkitSessionResourceLog
 from .whatsapp import Whatsapp
 from .whatsapp_contacts import WhatsappContacts
 
-MIXED_MODULES: list[type[MVTModule]] = [
+MIXED_MODULES = [
     Calls,
     ChromeFavicon,
     ChromeHistory,

tests/android/test_artifact_dumpsys_accessibility.py

@@ -23,7 +23,7 @@ class TestDumpsysAccessibilityArtifact:
         assert len(da.results) == 4
         assert da.results[0]["package_name"] == "com.android.settings"
         assert (
-            da.results[0]["component"]
+            da.results[0]["service"]
             == "com.android.settings/com.samsung.android.settings.development.gpuwatch.GPUWatchInterceptor"
         )
 
@@ -37,9 +37,7 @@ class TestDumpsysAccessibilityArtifact:
         da.parse(data)
         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
-        assert da.results[0]["installed"] is False
+        assert da.results[0]["service"] == "com.malware.service.malwareservice"
 
     def test_accessibility_service_alert(self):
         da = DumpsysAccessibilityArtifact()
@@ -54,22 +52,6 @@ class TestDumpsysAccessibilityArtifact:
         assert da.alertstore.alerts[0].level == AlertLevel.MEDIUM
         assert da.alertstore.alerts[0].event == da.results[0]
 
-    def test_same_component_is_kept_for_each_user(self):
-        da = DumpsysAccessibilityArtifact()
-        da.parse(
-            """User state[attributes:{id=0
-  installed services: {
-    0 : com.example/.Service
-  }
-User state[attributes:{id=10
-  installed services: {
-    0 : com.example/.Service
-  }
-"""
-        )
-
-        assert [result["user_id"] for result in da.results] == [0, 10]
-
     def test_ioc_check(self, indicator_file):
         da = DumpsysAccessibilityArtifact()
         file = get_artifact("android_data/dumpsys_accessibility.txt")

tests/android/test_artifact_dumpsys_adb.py

@@ -3,8 +3,6 @@
 # Use of this software is governed by the MVT License 1.1 that can be found at
 #   https://license.mvt.re/1.1/
 
-import base64
-
 from mvt.android.artifacts.dumpsys_adb import DumpsysADBArtifact
 from mvt.android.modules.bugreport.dumpsys_adb_state import DumpsysADBState
 from mvt.common.alerts import AlertLevel
@@ -13,17 +11,6 @@ from ..utils import get_artifact
 
 
 class TestDumpsysADBArtifact:
-    def test_parsing_binary_xml_recovers_key(self):
-        public_key = base64.b64encode(bytes(range(256)))
-        keystore = DumpsysADBArtifact().parse_binary_xml(
-            b"ABX\x00binary-key=" + public_key + b" user@host\x00lastConnection"
-        )
-
-        assert len(keystore) == 1
-        assert keystore[0]["key"] == public_key.decode()
-        assert keystore[0]["user"] == "user@host"
-        assert keystore[0]["last_connected"] is None
-
     def test_parsing(self):
         da_adb = DumpsysADBArtifact()
         file = get_artifact("android_data/dumpsys_adb.txt")

tests/android/test_artifact_dumpsys_appops.py

@@ -25,37 +25,10 @@ class TestDumpsysAppopsArtifact:
         assert da.results[0]["uid"] == "0"
         assert len(da.results[0]["permissions"]) == 1
         assert da.results[0]["permissions"][0]["name"] == "MANAGE_IPSEC_TUNNELS"
-        assert da.results[0]["permissions"][0]["mode"] == "allow"
+        assert da.results[0]["permissions"][0]["access"] == "allow"
         assert da.results[6]["package_name"] == "com.sec.factory.camera"
         assert len(da.results[6]["permissions"][1]["entries"]) == 1
         assert len(da.results[11]["permissions"]) == 4
-        wake_lock = next(
-            permission
-            for permission in da.results[11]["permissions"]
-            if permission["name"] == "WAKE_LOCK"
-        )
-        assert wake_lock["entries"][0]["duration"] == "+126ms"
-
-    def test_running_and_attribution_are_retained(self):
-        da = DumpsysAppopsArtifact()
-        da.parse(
-            """  Uid 0:
-    state=cch
-    Package com.example:
-      CAMERA (allow):
-        camera=[
-          Access: [fg-s] 2025-01-01 00:00:00.000 (-1s) duration=+2ms
-        ]
-      RECORD_AUDIO (allow):
-          Running start at: +3s
-"""
-        )
-
-        camera = da.results[0]["permissions"][0]["entries"][0]
-        running = da.results[0]["permissions"][1]["entries"][0]
-        assert camera["attribution"] == "camera"
-        assert running["event"] == "running"
-        assert running["relative_time"] == "+3s"
 
     def test_ioc_check(self, indicator_file):
         da = DumpsysAppopsArtifact()

tests/android/test_artifact_dumpsys_battery_daily.py

@@ -57,19 +57,18 @@ class TestDumpsysBatteryDailyArtifact:
         assert uninstall_alert.message == (
             "Detected uninstall of package com.example.removed (vers 0)"
         )
-        assert uninstall_alert.event_time == "2022-08-16 15:56:39"
+        assert uninstall_alert.event_time == "2022-08-16"
         assert uninstall_alert.event["package_name"] == "com.example.removed"
-        assert uninstall_alert.event["version_code"] == 0
-        assert uninstall_alert.event["action"] == "uninstall"
+        assert uninstall_alert.event["vers"] == "0"
 
         assert downgrade_alert.level == AlertLevel.MEDIUM
         assert downgrade_alert.message == (
             "Detected downgrade of package com.example.app from vers 10 to vers 9"
         )
-        assert downgrade_alert.event_time == "2022-08-17 15:56:39"
+        assert downgrade_alert.event_time == "2022-08-17"
         assert downgrade_alert.event["package_name"] == "com.example.app"
         assert downgrade_alert.event["action"] == "downgrade"
-        assert downgrade_alert.event["previous_version_code"] == 10
+        assert downgrade_alert.event["previous_vers"] == "10"
 
     def test_newest_first_update_is_not_reported_as_downgrade(self):
         dba = DumpsysBatteryDailyArtifact()
@@ -108,19 +107,7 @@ class TestDumpsysBatteryDailyArtifact:
         assert downgrade_alert.event_time == "2026-01-10"
         assert downgrade_alert.event["package_name"] == "com.example.app"
         assert downgrade_alert.event["action"] == "downgrade"
-        assert downgrade_alert.event["previous_version_code"] == 102
-
-    def test_duplicate_updates_retain_occurrence_count(self):
-        dba = DumpsysBatteryDailyArtifact()
-        dba.parse(
-            """  Daily from 2026-01-10-01-02-03 to 2026-01-11-04-05-06:
-      Update com.example.app vers=12
-      Update com.example.app vers=12
-"""
-        )
-
-        assert dba.results[0]["occurrences"] == 2
-        assert dba.results[0]["period_start"] == "2026-01-10 01:02:03"
+        assert downgrade_alert.event["previous_vers"] == "102"
 
     def test_reinstall_after_uninstall_is_not_reported_as_downgrade(self):
         dba = DumpsysBatteryDailyArtifact()

tests/android/test_artifact_dumpsys_battery_history.py

@@ -54,7 +54,6 @@ class TestDumpsysBatteryHistoryArtifact:
         assert len(dba.results) == 2
         assert dba.results[0] == {
             "time_elapsed": "07-15 20:27:39.431",
-            "timestamp": "1900-07-15 20:27:39.431000",
             "event": "start_job",
             "uid": "u0a123",
             "package_name": "com.example",
@@ -62,23 +61,3 @@ class TestDumpsysBatteryHistoryArtifact:
         }
         assert dba.results[1]["event"] == "end_job"
         assert dba.results[1]["uid"] == "u0a123"
-
-    def test_wake_lock_without_component_is_retained(self):
-        dba = DumpsysBatteryHistoryArtifact()
-        dba.parse(
-            "Battery History:\n"
-            "  0 (2) 100 RESET:TIME: 2025-09-05-01-04-52-139\n"
-            '  +1s (2) 100 +running +wake_lock=1000:"*alarm*:TIME_TICK"\n'
-            '  +2s (2) 100 +running +wake_lock=u0a1:"*walarm*:com.whatsapp.MessageHandler.LOGOUT_ACTION"\n'
-            "\n"
-        )
-
-        assert [record["event"] for record in dba.results] == ["wake", "wake"]
-        assert dba.results[0]["package_name"] is None
-        assert dba.results[1]["package_name"] == "com.whatsapp"
-
-    def test_decorated_sync_job_uses_component_package(self):
-        dba = DumpsysBatteryHistoryArtifact()
-        dba.parse('+1s (2) 100 +job=u0a1:"@SyncManager@gmail-ls/com.google:android"\n')
-
-        assert dba.results[0]["package_name"] == "com.google"

tests/android/test_artifact_dumpsys_dbinfo.py

@@ -53,32 +53,9 @@ Connection pool for /data/user/0/com.example/databases/current.db:
 
         assert dbi.results == [
             {
-                "timestamp": "07-15 20:27:39.431",
-                "pid": None,
+                "isodate": "07-15 20:27:39.431",
                 "action": "executeForCursorWindow",
-                "duration_ms": 1,
-                "status": "succeeded",
                 "sql": "SELECT 1",
                 "path": "/data/user/0/com.example/databases/current.db",
-                "pool_path": "/data/user/0/com.example/databases/current.db",
-                "connection_number": None,
-                "is_primary": None,
             }
         ]
-
-    def test_parses_operations_from_multiple_connections(self):
-        dbi = DumpsysDBInfoArtifact()
-        dbi.parse(
-            """Connection pool for /data/example.db:
-    Connection #0:
-      isPrimaryConnection: true
-      Most recently executed operations:
-        0: [2025-01-01 00:00:00.000] execute took 1ms - succeeded, sql="SELECT 1", path=/data/example.db
-    Connection #1:
-      isPrimaryConnection: false
-      Most recently executed operations:
-        0: [2025-01-01 00:00:01.000] execute took 2ms - succeeded, sql="SELECT 2", path=/data/example.db
-"""
-        )
-
-        assert [record["connection_number"] for record in dbi.results] == [0, 1]

tests/android/test_artifact_dumpsys_package_activities.py

@@ -21,16 +21,12 @@ class TestDumpsysPackageActivitiesArtifact:
 
         assert len(dpa.results) == 0
         dpa.parse(data)
-        assert len(dpa.results) == 10
-        assert dpa.results[0]["package_name"] == "com.samsung.android.messaging"
+        assert len(dpa.results) == 4
+        assert dpa.results[0]["package_name"] == "com.samsung.android.app.social"
         assert (
-            dpa.results[0]["component"]
-            == "com.samsung.android.messaging/.ui.RcsTransferContent"
+            dpa.results[0]["activity"]
+            == "com.samsung.android.app.social/.feed.FeedsActivity"
         )
-        assert {result["resolver_type"] for result in dpa.results} == {
-            "full_mime_type",
-            "non_data_action",
-        }
 
     def test_ioc_check(self, indicator_file):
         dpa = DumpsysPackageActivitiesArtifact()
@@ -45,4 +41,4 @@ class TestDumpsysPackageActivitiesArtifact:
         dpa.indicators = ind
         assert len(dpa.alertstore.alerts) == 0
         dpa.check_indicators()
-        assert len(dpa.alertstore.alerts) == 2
+        assert len(dpa.alertstore.alerts) == 1

tests/android/test_artifact_dumpsys_packages.py

@@ -25,11 +25,7 @@ class TestDumpsysPackagesArtifact:
             == "com.samsung.android.provider.filterprovider"
         )
         assert dpa.results[0]["version_name"] == "5.0.07"
-        assert dpa.results[0]["version_code"] == 500700000
-        assert dpa.results[0]["min_sdk"] == 28
-        assert dpa.results[0]["target_sdk"] == 28
-        assert dpa.results[0]["package_type"] == "active"
-        assert dpa.results[0]["users"][0]["user_id"] == 0
+        assert dpa.results[0]["first_install_time"] == "2008-12-31 16:00:00"
         assert dpa.results[0]["system"] is True
 
     def test_parsing_system_flag(self):
@@ -64,27 +60,42 @@ class TestDumpsysPackagesArtifact:
         dpa.check_indicators()
         assert len(dpa.alertstore.alerts) == 1
 
-    def test_hidden_packages_and_per_user_state(self):
-        dpa = DumpsysPackagesArtifact()
-        dpa.parse(
-            """Packages:
-  Package [com.example.active]:
-    appId=10001
-    versionCode=12 minSdk=29 targetSdk=35
-    User 0: installed=true hidden=false
-      firstInstallTime=2025-01-01 01:02:03
-    User 10: installed=false hidden=true
-      firstInstallTime=2025-01-02 01:02:03
-Hidden system packages:
-  Package [com.example.hidden]:
-    appId=10002
-    installerPackageName=null
+    def test_per_user_fields_use_primary_user(self):
+        details = DumpsysPackagesArtifact.parse_dumpsys_package_for_details(
+            """    User 0: installed=true
+      firstInstallTime=2024-01-10 09:19:39
+      runtime permissions:
+        android.permission.CAMERA: granted=true
+    User 95: installed=false
+      firstInstallTime=1970-01-01 01:00:00
+      runtime permissions:
+        android.permission.CAMERA: granted=false
+        android.permission.RECORD_AUDIO: granted=false
 """
         )
 
-        assert [record["package_type"] for record in dpa.results] == [
-            "active",
-            "hidden_system",
+        assert details["first_install_time"] == "2024-01-10 09:19:39"
+        runtime_permissions = [
+            permission
+            for permission in details["permissions"]
+            if permission["type"] == "runtime"
         ]
-        assert len(dpa.results[0]["users"]) == 2
-        assert dpa.results[1]["installer"] is None
+        assert runtime_permissions == [
+            {
+                "name": "android.permission.CAMERA",
+                "granted": True,
+                "type": "runtime",
+            }
+        ]
+
+    def test_per_user_fields_fall_back_when_user_zero_is_missing(self):
+        details = DumpsysPackagesArtifact.parse_dumpsys_package_for_details(
+            """    User 10: installed=true
+      firstInstallTime=2024-02-10 09:19:39
+      runtime permissions:
+        android.permission.CAMERA: granted=true
+"""
+        )
+
+        assert details["first_install_time"] == "2024-02-10 09:19:39"
+        assert details["permissions"][-1]["name"] == "android.permission.CAMERA"

tests/android/test_artifact_dumpsys_platform_compat.py

@@ -22,9 +22,6 @@ class TestDumpsysPlatformCompatArtifact:
         assert len(dbi.results) == 2
         assert dbi.results[0]["package_name"] == "org.torproject.torbrowser"
         assert dbi.results[1]["package_name"] == "org.article19.circulo.next"
-        assert dbi.results[0]["change_id"] == 168419799
-        assert dbi.results[0]["change_name"] == "DOWNSCALED"
-        assert dbi.results[0]["override_value"] is False
 
     def test_ioc_check(self, indicator_file):
         dbi = DumpsysPlatformCompatArtifact()

tests/android/test_artifact_dumpsys_receivers.py

@@ -19,14 +19,17 @@ class TestDumpsysReceiversArtifact:
 
         assert len(dr.results) == 0
         dr.parse(data)
-        assert len(dr.results) == 9
-        assert dr.results[0]["resolver_type"] == "full_mime_type"
-        storage_manager = next(
-            result
-            for result in dr.results
-            if result["key"] == "com.android.storagemanager.automatic.SHOW_NOTIFICATION"
+        assert len(dr.results) == 4
+        assert (
+            list(dr.results.keys())[0]
+            == "com.android.storagemanager.automatic.SHOW_NOTIFICATION"
+        )
+        assert (
+            dr.results["com.android.storagemanager.automatic.SHOW_NOTIFICATION"][0][
+                "package_name"
+            ]
+            == "com.android.storagemanager"
         )
-        assert storage_manager["package_name"] == "com.android.storagemanager"
 
     def test_parsing_misindented_action(self):
         dr = DumpsysReceiversArtifact()
@@ -41,8 +44,12 @@ Receiver Resolver Table:
 
         dr.parse(data)
 
-        assert dr.results[1]["key"] == "android.intent.action.MY_PACKAGE_REPLACED"
-        assert dr.results[1]["package_name"] == "com.psycatgames.nhiegame"
+        assert (
+            dr.results["android.intent.action.MY_PACKAGE_REPLACED"][0][
+                "package_name"
+            ]
+            == "com.psycatgames.nhiegame"
+        )
 
     def test_parsing_misindented_first_action(self):
         dr = DumpsysReceiversArtifact()
@@ -55,8 +62,12 @@ Receiver Resolver Table:
 
         dr.parse(data)
 
-        assert dr.results[0]["key"] == "android.intent.action.MY_PACKAGE_REPLACED"
-        assert dr.results[0]["package_name"] == "com.psycatgames.nhiegame"
+        assert (
+            dr.results["android.intent.action.MY_PACKAGE_REPLACED"][0][
+                "package_name"
+            ]
+            == "com.psycatgames.nhiegame"
+        )
 
     def test_ioc_check(self, indicator_file):
         dr = DumpsysReceiversArtifact()

tests/android/test_artifact_getprop.py

@@ -39,12 +39,3 @@ class TestGetPropArtifact:
         assert len(gp.alertstore.alerts) == 0
         gp.check_indicators()
         assert len(gp.alertstore.alerts) == 1
-
-    def test_empty_values_and_invalid_lines(self):
-        gp = GetProp()
-        gp.parse("[empty]: []\n[valid]: [value]\n0\n[broken]: [value")
-
-        assert gp.results == [
-            {"name": "empty", "value": ""},
-            {"name": "valid", "value": "value"},
-        ]

tests/android/test_artifact_processes.py

@@ -20,7 +20,7 @@ class TestProcessesArtifact:
         assert len(p.results) == 0
         p.parse(data)
         assert len(p.results) == 17
-        assert p.results[0]["command"] == "init"
+        assert p.results[0]["proc_name"] == "init"
 
     def test_ioc_check(self, indicator_file):
         p = Processes()
@@ -36,14 +36,3 @@ class TestProcessesArtifact:
         assert len(p.alertstore.alerts) == 0
         p.check_indicators()
         assert len(p.alertstore.alerts) == 1
-
-    def test_bugreport_thread_columns(self):
-        p = Processes()
-        p.parse(
-            "LABEL USER PID TID PPID VSZ RSS WCHAN ADDR S PRI NI RTPRIO SCH PCY TIME CMD\n"
-            "u:r:init:s0 root 1 2 0 100 20 0 0 S 19 0 - 0 fg 00:00:01 init\n"
-        )
-
-        assert p.results[0]["label"] == "u:r:init:s0"
-        assert p.results[0]["tid"] == 2
-        assert p.results[0]["command"] == "init"

tests/android/test_artifact_tombstones.py

@@ -128,5 +128,8 @@ class TestTombstoneCrashArtifact:
         assert tombstone_result.get("pid") == 25541
         assert tombstone_result.get("process_name") == "mtk.ape.decoder"
 
-        # Tombstones include an explicit offset, so normalize them to UTC.
-        assert tombstone_result.get("timestamp") == "2023-04-12 10:32:40.518290"
+        # With Android logs we want to keep timestamps as device local time for consistency.
+        # We often don't know the time offset for a log entry and so can't convert everything to UTC.
+        # MVT should output the local time only:
+        # So original 2023-04-12 12:32:40.518290770+0200 -> 2023-04-12 12:32:40.000000
+        assert tombstone_result.get("timestamp") == "2023-04-12 12:32:40.518290"

tests/android_androidqf/test_mounts.py

@@ -13,19 +13,6 @@ from ..utils import get_android_androidqf, list_files
 
 
 class TestAndroidqfMountsArtifact:
-    def test_parse_proc_mountinfo(self):
-        from mvt.android.artifacts.mounts import Mounts as MountsArtifact
-
-        results = MountsArtifact.parse_mountinfo(
-            "41 40 254:13 / / ro,relatime shared:1 - erofs /dev/block/dm-13 ro,seclabel\n",
-            123,
-        )
-
-        assert results[0]["mount_id"] == 41
-        assert results[0]["mount_point"] == "/"
-        assert results[0]["filesystem_type"] == "erofs"
-        assert results[0]["process_ids"] == [123]
-
     def test_parse_mounts_token_checks(self):
         """
         Test the artifact-level `parse` method using tolerant token checks.

tests/android_androidqf/test_settings.py

@@ -6,27 +6,12 @@
 from pathlib import Path
 
 from mvt.android.modules.androidqf.aqf_settings import AQFSettings
-from mvt.android.artifacts.settings import Settings
 from mvt.common.module import run_module
 
 from ..utils import get_android_androidqf, list_files
 
 
 class TestSettingsModule:
-    def test_bugreport_settings_format(self):
-        settings = Settings()
-        settings.parse(
-            "GLOBAL SETTINGS (user 0)\n"
-            "_id:1 name:adb_wifi_enabled pkg:android value:0 default:0 defaultSystemSet:true\n"
-            "SECURE SETTINGS (user 10)\n"
-            "_id:2 name:accessibility_enabled pkg:android value:1\n"
-        )
-
-        assert settings.results == {
-            "global:user_0": {"adb_wifi_enabled": "0"},
-            "secure:user_10": {"accessibility_enabled": "1"},
-        }
-
     def test_parsing(self):
         data_path = get_android_androidqf()
         m = AQFSettings(target_path=data_path)

tests/android_bugreport/test_bugreport.py

@@ -54,11 +54,10 @@ class TestBugreportAnalysis:
             == "com.samsung.android.provider.filterprovider"
         )
         assert m.results[1]["package_name"] == "com.instagram.android"
-        assert m.results[0]["installer"] is None
+        assert m.results[0]["installer"] == ""
         assert m.results[1]["installer"] == "com.android.vending"
         assert len(m.results[0]["permissions"]) == 4
-        assert len(m.results[1]["permissions"]) == 20
-        assert len(m.results[1]["users"][0]["permissions"]) == 19
+        assert len(m.results[1]["permissions"]) == 32
 
     def test_getprop_module(self):
         m = self.launch_bug_report_module(DumpsysGetProp)
@@ -67,34 +66,29 @@ class TestBugreportAnalysis:
     def test_receivers_match_exact_package_name(self, indicators_factory):
         intent = "android.intent.action.PHONE_STATE"
         false_positive = {
-            "resolver_type": "non_data_action",
-            "key": intent,
             "package_name": "com.android.phone",
-            "component": (
+            "receiver": (
                 "com.android.phone/"
                 "com.android.services.telephony.sip.SipIncomingCallReceiver"
             ),
-            "filter_count": 1,
         }
         malicious_receiver = {
-            "resolver_type": "non_data_action",
-            "key": intent,
             "package_name": "com.android.services",
-            "component": "com.android.services/com.example.SomeReceiver",
-            "filter_count": 1,
+            "receiver": "com.android.services/com.example.SomeReceiver",
         }
-        module = DumpsysReceivers(results=[false_positive, malicious_receiver])
+        module = DumpsysReceivers(
+            results={intent: [false_positive, malicious_receiver]}
+        )
         module.indicators = indicators_factory(app_ids=["com.android.services"])
 
         module.check_indicators()
 
         assert len(module.alertstore.alerts) == 1
         alert = module.alertstore.alerts[0]
-        assert alert.event == malicious_receiver
+        assert alert.event == {intent: malicious_receiver}
         assert alert.matched_indicator.value == "com.android.services"
 
     def test_tombstones_modules(self):
         m = self.launch_bug_report_module(Tombstones)
         assert len(m.results) == 2
         assert m.results[1]["pid"] == 3559
-        assert m.results[0]["sources"]["text"]["parsed"] is True