Fix VMRay missing process data

CHANGELOG.md

@@ -21,6 +21,7 @@
 
 - use Python 3.12 to build extra standalone build on Linux #2383 @williballenthin
 - bump minimum Python version to 3.8.1 to satisfy uv #2387 @williballenthin
+- collect more process information from flog.xml #2394 @mr-tz
 
 ### capa explorer IDA Pro plugin
 

capa/features/extractors/vmray/__init__.py

@@ -45,8 +45,8 @@ def __init__(self, zipfile_path: Path):
         self.exports: Dict[int, str] = {}
         self.imports: Dict[int, Tuple[str, str]] = {}
         self.sections: Dict[int, str] = {}
-        self.process_ids: Dict[int, int] = {}
-        self.process_threads: Dict[int, List[int]] = defaultdict(list)
+        self.os_pid_by_monitor_id: Dict[int, int] = {}
+        self.tids_by_pid: Dict[int, List[int]] = defaultdict(list)
         self.process_calls: Dict[int, Dict[int, List[FunctionCall]]] = defaultdict(lambda: defaultdict(list))
         self.base_address: int
 
@@ -131,8 +131,15 @@ def _compute_sections(self):
     def _compute_process_ids(self):
         for process in self.sv2.processes.values():
             # we expect VMRay's monitor IDs to be unique, but OS PIDs may be reused
-            assert process.monitor_id not in self.process_ids.keys()
-            self.process_ids[process.monitor_id] = process.os_pid
+            assert process.monitor_id not in self.os_pid_by_monitor_id.keys()
+            self.os_pid_by_monitor_id[process.monitor_id] = process.os_pid
+
+        # not all processes may get an ID, get missing data from flog.xml, see #2394
+        for monitor_process in self.flog.analysis.monitor_processes:
+            if monitor_process.process_id in self.os_pid_by_monitor_id:
+                assert self.os_pid_by_monitor_id[monitor_process.process_id] == monitor_process.os_pid
+            else:
+                self.os_pid_by_monitor_id[monitor_process.process_id] = monitor_process.os_pid
 
     def _compute_process_threads(self):
         # logs/flog.xml appears to be the only file that contains thread-related data
@@ -144,8 +151,8 @@ def _compute_process_threads(self):
             assert isinstance(pid, int)
             assert isinstance(tid, int)
 
-            if tid not in self.process_threads[pid]:
-                self.process_threads[pid].append(tid)
+            if tid not in self.tids_by_pid[pid]:
+                self.tids_by_pid[pid].append(tid)
 
     def _compute_process_calls(self):
         for function_call in self.flog.analysis.function_calls:
@@ -158,4 +165,4 @@ def _compute_process_calls(self):
             self.process_calls[pid][tid].append(function_call)
 
     def get_process_os_pid(self, monitor_id: int) -> int:
-        return self.process_ids[monitor_id]
+        return self.os_pid_by_monitor_id[monitor_id]

capa/features/extractors/vmray/extractor.py

@@ -80,7 +80,7 @@ def get_process_name(self, ph) -> str:
         return process.image_name
 
     def get_threads(self, ph: ProcessHandle) -> Iterator[ThreadHandle]:
-        for thread in self.analysis.process_threads[ph.address.pid]:
+        for thread in self.analysis.tids_by_pid[ph.address.pid]:
             address: ThreadAddress = ThreadAddress(process=ph.address, tid=thread)
             yield ThreadHandle(address=address, inner={})
 

capa/features/extractors/vmray/models.py

@@ -87,7 +87,7 @@ class Param(BaseModel):
     deref: Optional[ParamDeref] = None
 
 
-def validate_param_list(value: Union[List[Param], Param]) -> List[Param]:
+def validate_ensure_is_list(value: Union[List[Param], Param]) -> List[Param]:
     if isinstance(value, list):
         return value
     else:
@@ -97,7 +97,7 @@ def validate_param_list(value: Union[List[Param], Param]) -> List[Param]:
 # params may be stored as a list of Param or a single Param so we convert
 # the input value to Python list type before the inner validation (List[Param])
 # is called
-ParamList = Annotated[List[Param], BeforeValidator(validate_param_list)]
+ParamList = Annotated[List[Param], BeforeValidator(validate_ensure_is_list)]
 
 
 class Params(BaseModel):
@@ -137,12 +137,37 @@ class FunctionReturn(BaseModel):
     from_addr: HexInt = Field(alias="from")
 
 
+class MonitorProcess(BaseModel):
+    ts: HexInt
+    process_id: int
+    image_name: str
+    filename: str
+    # page_root: HexInt
+    os_pid: HexInt
+    # os_integrity_level: HexInt
+    # os_privileges: HexInt
+    monitor_reason: str
+    parent_id: int
+    os_parent_pid: HexInt
+    # cmd_line: str
+    # cur_dir: str
+    # os_username: str
+    # bitness: int
+    # os_groups: str
+
+
+# handle if there's only single entries, but the model expects a list
+MonitorProcessList = Annotated[List[MonitorProcess], BeforeValidator(validate_ensure_is_list)]
+FunctionCallList = Annotated[List[FunctionCall], BeforeValidator(validate_ensure_is_list)]
+
+
 class Analysis(BaseModel):
     log_version: str  # tested 2
     analyzer_version: str  # tested 2024.2.1
     # analysis_date: str
 
-    function_calls: List[FunctionCall] = Field(alias="fncall", default=[])
+    monitor_processes: MonitorProcessList = Field(alias="monitor_process", default=[])
+    function_calls: FunctionCallList = Field(alias="fncall", default=[])
     # function_returns: List[FunctionReturn] = Field(alias="fnret", default=[])
 
 

tests/fixtures.py

@@ -431,6 +431,14 @@ def get_data_path_by_name(name) -> Path:
             / "vmray"
             / "93b2d1840566f45fab674ebc79a9d19c88993bcb645e0357f3cb584d16e7c795_min_archive.zip"
         )
+    elif name.startswith("2f8a79-vmray"):
+        return (
+            CD
+            / "data"
+            / "dynamic"
+            / "vmray"
+            / "2f8a79b12a7a989ac7e5f6ec65050036588a92e65aeb6841e08dc228ff0e21b4_min_archive.zip"
+        )
     elif name.startswith("ea2876"):
         return CD / "data" / "ea2876e9175410b6f6719f80ee44b9553960758c7d0f7bed73c0fe9a78d8e669.dll_"
     elif name.startswith("1038a2"):

tests/test_vmray_features.py

@@ -87,3 +87,10 @@ def test_vmray_features(sample, scope, feature, expected):
 )
 def test_vmray_feature_counts(sample, scope, feature, expected):
     fixtures.do_test_feature_count(fixtures.get_vmray_extractor, sample, scope, feature, expected)
+
+
+def test_vmray_processes():
+    # see #2394
+    path = fixtures.get_data_path_by_name("2f8a79-vmray")
+    vmre = fixtures.get_vmray_extractor(path)
+    assert len(vmre.analysis.os_pid_by_monitor_id) == 9