Add support for parsing Protobuf-formatted artifacts

src/mvt/android/artifacts/tombstone_crashes.py

@@ -10,7 +10,7 @@
 import betterproto2
 from dateutil import parser
 
-from mvt.android.parsers.proto.tombstone import Tombstone
+from mvt.android.parsers.protobuf_parsers import parse_tombstone_record
 from mvt.common.module_types import ModuleAtomicResult, ModuleSerializedResult
 from mvt.common.utils import convert_datetime_to_iso
 
@@ -128,7 +128,7 @@ def parse_protobuf(
         self, file_name: str, file_timestamp: datetime.datetime, data: bytes
     ) -> None:
         """Parse Android tombstone crash files from a protobuf object."""
-        tombstone_pb = Tombstone().parse(data)
+        tombstone_pb = parse_tombstone_record(data)
         tombstone_dict = tombstone_pb.to_dict(
             casing=betterproto2.Casing.SNAKE, include_default_values=True
         )

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

@@ -14,6 +14,7 @@
 from typing import Optional
 
 from mvt.android.modules.androidqf.base import AndroidQFModule
+from mvt.android.parsers.protobuf_parsers import parse_files_records
 from mvt.common.module_types import (
     ModuleAtomicResult,
     ModuleResults,
@@ -125,6 +126,7 @@ def run(self) -> None:
             self.log.warning("Unable to determine device timezone, using UTC")
             device_timezone = zoneinfo.ZoneInfo("UTC")
 
+        data = []
         for file in self._get_files_by_pattern("*/files.json"):
             rawdata = self._get_file_content(file).decode("utf-8", errors="ignore")
             try:
@@ -136,24 +138,31 @@ def run(self) -> None:
                         continue
                     data.append(json.loads(line))
 
-            for file_data in data:
-                for ts in ["access_time", "changed_time", "modified_time"]:
-                    if ts in file_data:
-                        utc_timestamp = datetime.datetime.fromtimestamp(
-                            file_data[ts], tz=datetime.timezone.utc
-                        )
-                        # Convert the UTC timestamp to local time on Android device's local timezone
-                        local_timestamp = utc_timestamp.astimezone(device_timezone)
-
-                        # Preserve the device-local wall-clock time while using
-                        # the project-wide ISO conversion helper.
-                        local_timestamp = local_timestamp.replace(
-                            tzinfo=datetime.timezone.utc
-                        )
-                        file_data[ts] = convert_datetime_to_iso(local_timestamp)
-
-                self.results.append(file_data)
-
-            break  # Only process the first matching file
+        if data == []:
+            for file in self._get_files_by_pattern("*/files.pb"):
+                try:
+                    data = parse_files_records(self._get_file_content(file))
+                except ValueError as exc:
+                    self.log.error("Failed to parse files.pb: %s", exc)
+                    return
+                break
+
+        for file_data in data:
+            for ts in ["access_time", "changed_time", "modified_time"]:
+                if ts in file_data:
+                    utc_timestamp = datetime.datetime.fromtimestamp(
+                        file_data[ts], tz=datetime.timezone.utc
+                    )
+                    # Convert the UTC timestamp to local time on Android device's local timezone
+                    local_timestamp = utc_timestamp.astimezone(device_timezone)
+
+                    # Preserve the device-local wall-clock time while using
+                    # the project-wide ISO conversion helper.
+                    local_timestamp = local_timestamp.replace(
+                        tzinfo=datetime.timezone.utc
+                    )
+                    file_data[ts] = convert_datetime_to_iso(local_timestamp)
+
+            self.results.append(file_data)
 
         self.log.info("Found a total of %d files", len(self.results))

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

@@ -10,6 +10,7 @@
 
 from rich.progress import track
 
+from mvt.android.parsers.protobuf_parsers import parse_packages_records
 from mvt.android.utils import (
     BROWSER_INSTALLERS,
     PLAY_STORE_INSTALLERS,
@@ -190,11 +191,18 @@ def check_virustotal(self, delay: int = 0) -> None:
 
     def run(self) -> None:
         packages = self._get_files_by_pattern("*/packages.json")
-        if not packages:
-            self.log.error(
-                "packages.json file not found in this androidqf bundle. Possibly malformed?"
-            )
+        if packages:
+            self.results = json.loads(self._get_file_content(packages[0]))
+            self.log.info("Found %d packages in packages.json", len(self.results))
+            return
+
+        packages = self._get_files_by_pattern("*/packages.pb")
+        if packages:
+            self.results = parse_packages_records(self._get_file_content(packages[0]))
+            self.log.info("Found %d packages in packages.pb", len(self.results))
             return
 
-        self.results = json.loads(self._get_file_content(packages[0]))
-        self.log.info("Found %d packages in packages.json", len(self.results))
+        self.log.error(
+            "packages.json or packages.pb file not found in this androidqf bundle. Possibly malformed?"
+        )
+        return

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

@@ -8,6 +8,7 @@
 from typing import Optional
 
 from mvt.android.artifacts.mounts import Mounts as MountsArtifact
+from mvt.android.parsers.protobuf_parsers import parse_string_records
 
 from .base import AndroidQFModule
 
@@ -34,6 +35,14 @@ def __init__(
         )
         self.results: list = [] if results is None else results
 
+    def _load_json(self, file: str) -> list[str]:
+        data = self._get_file_content(file).decode("utf-8", errors="replace")
+        return json.loads(data)
+
+    def _load_pb(self, file: str) -> list[str]:
+        data = self._get_file_content(file)
+        return parse_string_records(data)
+
     def run(self) -> None:
         """
         Run the mounts analysis module.
@@ -42,30 +51,37 @@ def run(self) -> None:
         and analyzes them for suspicious configurations, particularly focusing
         on detecting root access indicators like /system mounted as read-write.
         """
-        mount_files = self._get_files_by_pattern("*/mounts.json")
 
-        if not mount_files:
-            self.log.info("No mount information file found")
-            return
+        mount_data = []
 
-        self.log.info("Found mount information file: %s", mount_files[0])
+        mount_files = self._get_files_by_pattern("*/mounts.json")
+        if mount_files:
+            try:
+                mount_data = self._load_json(mount_files[0])
+                self.log.info("Found mount information file: %s", mount_files[0])
+            except Exception as exc:
+                self.log.error("Failed to parse JSON mount information: %s", exc)
+                return
 
-        try:
-            data = self._get_file_content(mount_files[0]).decode(
-                "utf-8", errors="replace"
-            )
-        except Exception as exc:
-            self.log.error("Failed to read mount information file: %s", exc)
+        mount_files = self._get_files_by_pattern("*/mounts.pb")
+        if len(mount_data) == 0 and mount_files:
+            try:
+                mount_data = self._load_pb(mount_files[0])
+                self.log.info("Found mount information file: %s", mount_files[0])
+            except Exception as exc:
+                self.log.error("Failed to parse Protobuf mount information: %s", exc)
+                return
+
+        if len(mount_data) == 0:
+            self.log.info("No mount information file found")
             return
 
         # Parse the mount data
         try:
-            json_data = json.loads(data)
-
-            if isinstance(json_data, list):
+            if isinstance(mount_data, list):
                 # AndroidQF format: array of strings like
                 # "/dev/block/dm-12 on / type ext4 (ro,seclabel,noatime)"
-                mount_content = "\n".join(json_data)
+                mount_content = "\n".join(mount_data)
             else:
                 self.log.error("Expected mounts.json to contain a list of mount lines")
                 return

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

@@ -7,6 +7,8 @@
 import logging
 from typing import Optional
 
+from mvt.android.parsers.protobuf_parsers import parse_string_records
+
 from .base import AndroidQFModule
 
 
@@ -58,26 +60,42 @@ def check_indicators(self) -> None:
                 len(self.results),
             )
 
-    def run(self) -> None:
-        """Run the root binaries analysis."""
-        root_binaries_files = self._get_files_by_pattern("*/root_binaries.json")
+    def _load_json(self, file: str) -> list[str]:
+        data = self._get_file_content(file).decode("utf-8", errors="ignore")
+        return json.loads(data)
 
-        if not root_binaries_files:
-            self.log.info("No root_binaries.json file found")
-            return
+    def _load_pb(self, file: str) -> list[str]:
+        data = self._get_file_content(file)
+        return parse_string_records(data)
 
-        rawdata = self._get_file_content(root_binaries_files[0]).decode(
-            "utf-8", errors="ignore"
-        )
+    def run(self) -> None:
+        """Run the root binaries analysis."""
+        root_binary_paths = []
 
-        try:
-            root_binary_paths = json.loads(rawdata)
-        except json.JSONDecodeError as e:
-            self.log.error("Failed to parse root_binaries.json: %s", e)
+        root_binaries_files = self._get_files_by_pattern("*/root_binaries.json")
+        if root_binaries_files:
+            try:
+                root_binary_paths = self._load_json(root_binaries_files[0])
+                self.log.info("Found root_binaries.json file: %s", root_binaries_files[0])
+            except Exception as exc:
+                self.log.error("Failed to parse JSON root_binaries.json: %s", exc)
+                return
+
+        root_binaries_files = self._get_files_by_pattern("*/root_binaries.pb")
+        if len(root_binary_paths) == 0 and root_binaries_files:
+            try:
+                root_binary_paths = self._load_pb(root_binaries_files[0])
+                self.log.info("Found root_binaries.pb file: %s", root_binaries_files[0])
+            except Exception as exc:
+                self.log.error("Failed to parse Protobuf root_binaries.pb: %s", exc)
+                return
+
+        if len(root_binary_paths) == 0:
+            self.log.info("No root_binaries file found")
             return
 
         if not isinstance(root_binary_paths, list):
-            self.log.error("Expected root_binaries.json to contain a list of paths")
+            self.log.error("Expected root_binaries.json or root_binaries.pb to contain a list of paths")
             return
 
         # Known root binary names that might be found and their descriptions