Add bugreport process table parser

src/mvt/android/artifacts/processes.py

@@ -1,72 +1,91 @@
 # Mobile Verification Toolkit (MVT)
-# Copyright (c) 2021-2023 The MVT Authors.
+# 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/
 
 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:
-        for line in entry.splitlines()[1:]:
-            proc = line.split()
+        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
 
-            # Skip empty lines
-            if len(proc) == 0:
+        for line in lines[1:]:
+            values = line.split(None, len(headers) - 1)
+            if len(values) != len(headers):
                 continue
-
-            # 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,
-                }
-            )
+            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)
 
     def check_indicators(self) -> None:
         if not self.indicators:
             return
-
         for result in self.results:
-            proc_name = result.get("proc_name", "")
-            if not proc_name:
+            command = result.get("command", "")
+            if not isinstance(command, str):
                 continue
-
-            # Skipping this process because of false positives.
-            if result["proc_name"] == "gatekeeperd":
+            process_name = command.rsplit("/", 1)[-1]
+            if not process_name or process_name == "gatekeeperd":
                 continue
-
-            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
-                )
+            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

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

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

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]["proc_name"] == "init"
+        assert p.results[0]["command"] == "init"
 
     def test_ioc_check(self, indicator_file):
         p = Processes()
@@ -36,3 +36,14 @@ 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"