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

Code before:
                    self.imports[api.address] = (module.dll, api.api.name)

    def _compute_sections(self):
        assert self.sample_file_static_data is not None
        if self.sample_file_static_data.pe:
            for pefile_section in self.sample_file_static_data.pe.sections:
                self.sections[pefile_section.virtual_address] = pefile_section.name
        elif self.sample_file_static_data.elf:
            for elffile_section in self.sample_file_static_data.elf.sections:
                self.sections[elffile_section.header.sh_addr] = elffile_section.header.sh_name

    def _compute_monitor_processes(self):
        for process in self.sv2.processes.values():
            # we expect monitor IDs to be unique
            assert process.monitor_id not in self.monitor_processes

            ppid: int = (
                self.sv2.processes[process.ref_parent_process.path[1]].os_pid if process.ref_parent_process else 0
            )
            self.monitor_processes[process.monitor_id] = VMRayMonitorProcess(
                process.os_pid,
                ppid,
                process.monitor_id,
                process.image_name,
                sanitize_string(process.filename),
                sanitize_string(process.cmd_line),
            )

        # not all processes are recorded in SummaryV2.json, get missing data from flog.xml, see #2394
        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.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
                assert self.monitor_processes[monitor_process.process_id] == 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(

Review comment (security):
Is there a reason not to handle this via the `SanitizedString` as is done with `MonitorProcess`?

Code after:
            for elffile_section in self.sample_file_static_data.elf.sections:
                self.sections[elffile_section.header.sh_addr] = elffile_section.header.sh_name

    def _compute_monitor_processes(self):
        for process in self.sv2.processes.values():
            # we expect monitor IDs to be unique
            assert process.monitor_id not in self.monitor_processes

            ppid: int = (
                self.sv2.processes[process.ref_parent_process.path[1]].os_pid if process.ref_parent_process else 0
            )
            self.monitor_processes[process.monitor_id] = VMRayMonitorProcess(
                process.os_pid,
                ppid,
                process.monitor_id,
                process.image_name,
                process.filename,
                process.cmd_line,
            )

        # not all processes are recorded in SummaryV2.json, get missing data from flog.xml, see #2394
        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.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
                assert self.monitor_processes[monitor_process.process_id] == 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)