Fix Android bugreport parser coverage

src/mvt/android/artifacts/dumpsys_accessibility.py

@@ -4,6 +4,7 @@
 #   https://license.mvt.re/1.1/
 
 import re
+from typing import Any
 
 from .artifact import AndroidArtifact
 
@@ -20,10 +21,10 @@ def check_indicators(self) -> None:
                     continue
 
             self.alertstore.medium(
-                f'Found accessibility service: "{result["service"]}"',
+                f'Found accessibility service: "{result["component"]}"',
                 "",
                 result,
-                )
+            )
 
     def parse(self, content: str) -> None:
         """
@@ -33,41 +34,69 @@ def parse(self, content: str) -> None:
         :param content: content of the accessibility section (string)
         """
 
-        # "Old" syntax
-        in_services = False
+        self.results: list[dict[str, Any]] = []
+        services: dict[tuple[int | None, str], dict] = {}
+        user_id: int | None = None
+        state: str | None = None
+
         for line in content.splitlines():
-            if line.strip().startswith("installed services:"):
-                in_services = True
+            user_match = re.search(r"attributes:\{id=(\d+)", line)
+            if user_match:
+                user_id = int(user_match.group(1))
+
+            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)
+                    )
+                    service[self._state_field(state)] = True
+                    service["accessibility_tool"] = "(A11yTool)" in inline
                 continue
 
-            if not in_services:
+            if not state:
                 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,
-                }
+            if stripped == "}" or stripped.startswith("AccessibilityInputFilter"):
+                state = None
+                continue
+            component_match = re.search(
+                r"(?:\d+\s*:\s*)?([\w.$-]+/[\w.$-]+)(?:\s+\(A11yTool\))?",
+                stripped,
             )
-
-        # "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():
-            if line.strip().startswith("Enabled services:"):
-                matches = re.finditer(r"{([^{]+?)}", line)
-
-                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}
-                    )
+            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,6 +6,7 @@
 import base64
 import binascii
 import hashlib
+import re
 
 from .artifact import AndroidArtifact
 
@@ -98,6 +99,34 @@ def parse_xml(self, xml_data):
 
         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:
@@ -118,7 +147,7 @@ def calculate_key_info(user_key: bytes) -> dict:
         return {
             "user": user.decode("utf-8"),
             "fingerprint": key_fingerprint_colon,
-            "key": key_base64,
+            "key": key_base64.decode("ascii", errors="replace"),
         }
 
     def check_indicators(self) -> None:
@@ -175,12 +204,23 @@ def parse(self, content: bytes) -> None:
         # 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", []):