Repository: mandiant/capa
File: capa/features/extractors/vmray/__init__.py (Python)

Code before:
        for monitor_process in self.flog.analysis.monitor_processes:
            vmray_monitor_process: VMRayMonitorProcess = VMRayMonitorProcess(
                monitor_process.os_pid,
                monitor_process.os_parent_pid,
                monitor_process.process_id,
                monitor_process.parent_id,
                monitor_process.image_name,
                monitor_process.filename,
                monitor_process.cmd_line,
            )

            if monitor_process.process_id not in self.monitor_processes:
                self.monitor_processes[monitor_process.process_id] = vmray_monitor_process
            else:
                # we expect monitor processes recorded in both SummaryV2.json and flog.xml to equal
                # to ensure this, we compare the pid, monitor_id, and origin_monitor_id
                # for the other fields we've observed cases with slight deviations, e.g.,
                # the ppid for a process in flog.xml is not set correctly, all other data is equal
                sv2p = self.monitor_processes[monitor_process.process_id]
                assert (sv2p.pid, sv2p.monitor_id, sv2p.origin_monitor_id) == (
                    vmray_monitor_process.pid,
                    vmray_monitor_process.monitor_id,
                    vmray_monitor_process.origin_monitor_id,
                )
                if self.monitor_processes[monitor_process.process_id] != vmray_monitor_process:
                    logger.debug("processes differ: %s (sv2) vs. %s (flog)", sv2p, vmray_monitor_process)

    def _compute_monitor_threads(self):
        for monitor_thread in self.flog.analysis.monitor_threads:
            # we expect monitor IDs to be unique
            assert monitor_thread.thread_id not in self.monitor_threads

            self.monitor_threads[monitor_thread.thread_id] = VMRayMonitorThread(
                monitor_thread.os_tid, monitor_thread.thread_id, monitor_thread.process_id
            )

            # we expect each monitor thread ID to be unique for its associated monitor process ID e.g. monitor
            # thread ID 10 should not be captured twice for monitor process ID 1
            assert monitor_thread.thread_id not in self.monitor_threads_by_monitor_process[monitor_thread.thread_id]

            self.monitor_threads_by_monitor_process[monitor_thread.process_id].append(monitor_thread.thread_id)

    def _compute_monitor_process_calls(self):
        for function_call in self.flog.analysis.function_calls:
            self.monitor_process_calls[function_call.process_id][function_call.thread_id].append(function_call)


Review comment (suggestion):
thoughts on being even more lenient here and not asserting this but just reporting it?
I'm encountering more inconsistencies between the two files, e.g., the `monitor_id` not being set

Code after:
            vmray_monitor_process: VMRayMonitorProcess = VMRayMonitorProcess(
                monitor_process.os_pid,
                monitor_process.os_parent_pid,
                monitor_process.process_id,
                monitor_process.parent_id,
                monitor_process.image_name,
                monitor_process.filename,
                monitor_process.cmd_line,
            )

            if monitor_process.process_id not in self.monitor_processes:
                self.monitor_processes[monitor_process.process_id] = vmray_monitor_process
            else:
                # we expect monitor processes recorded in both SummaryV2.json and flog.xml to equal
                # to ensure this, we compare the pid, monitor_id, and origin_monitor_id
                # for the other fields we've observed cases with slight deviations, e.g.,
                # the ppid for a process in flog.xml is not set correctly, all other data is equal
                sv2p = self.monitor_processes[monitor_process.process_id]
                if self.monitor_processes[monitor_process.process_id] != vmray_monitor_process:
                    logger.debug("processes differ: %s (sv2) vs. %s (flog)", sv2p, vmray_monitor_process)

                assert (sv2p.pid, sv2p.monitor_id, sv2p.origin_monitor_id) == (
                    vmray_monitor_process.pid,
                    vmray_monitor_process.monitor_id,
                    vmray_monitor_process.origin_monitor_id,
                )

    def _compute_monitor_threads(self):
        for monitor_thread in self.flog.analysis.monitor_threads:
            # we expect monitor IDs to be unique
            assert monitor_thread.thread_id not in self.monitor_threads

            self.monitor_threads[monitor_thread.thread_id] = VMRayMonitorThread(
                monitor_thread.os_tid, monitor_thread.thread_id, monitor_thread.process_id
            )

            # we expect each monitor thread ID to be unique for its associated monitor process ID e.g. monitor
            # thread ID 10 should not be captured twice for monitor process ID 1
            assert monitor_thread.thread_id not in self.monitor_threads_by_monitor_process[monitor_thread.thread_id]

            self.monitor_threads_by_monitor_process[monitor_thread.process_id].append(monitor_thread.thread_id)

    def _compute_monitor_process_calls(self):
        for function_call in self.flog.analysis.function_calls:
            self.monitor_process_calls[function_call.process_id][function_call.thread_id].append(function_call)