Method _worker

bbot/modules/base.py:717–793

The core worker loop for the module, responsible for handling events from the incoming event queue. This method is a coroutine and is run asynchronously. Multiple instances can run simultaneously based on the 'module_threads' configuration. The worker dequeues events from '

(self)

        return self, status, str(msg)

    async def _worker(self):
        """
        The core worker loop for the module, responsible for handling events from the incoming event queue.

        This method is a coroutine and is run asynchronously. Multiple instances can run simultaneously based on
        the 'module_threads' configuration. The worker dequeues events from 'incoming_event_queue', performs
        necessary prechecks, and passes the event to the appropriate handler function.

        Args:
            None

        Returns:
            None

        Raises:
            asyncio.CancelledError: If the worker is cancelled during its operation.

        Notes:
            - The worker is sensitive to the 'stopping' flag of the scan. It will terminate if this flag is set.
            - The worker handles backpressure by pausing when the outgoing event queue is full.
            - Batch processing is supported and is activated when 'batch_size' > 1.
            - Each event is subject to a post-check via '_event_postcheck()' to decide whether it should be handled.
            - Special 'FINISHED' events trigger the 'finish()' method of the module.
        """
        async with self.scan._acatch(context=self._worker, unhandled_is_critical=True):
            try:
                while not self.scan.stopping and not self.errored:
                    # if batch wasn't big enough, we wait for the next event before continuing
                    if self.batch_size > 1:
                        submitted = await self._handle_batch()
                        if not submitted:
                            async with self.event_received:
                                await self.event_received.wait()

                    else:
                        try:
                            if self.incoming_event_queue is not False:
                                event = await self.incoming_event_queue.get()
                            else:
                                self.debug("Event queue is in bad state")
                                break
                        except asyncio.queues.QueueEmpty:
                            continue
                        self.debug(f"Got {event} from {getattr(event, 'module', 'unknown_module')}")
                        async with self._task_counter.count(f"event_postcheck({event})"):
                            acceptable, reason = await self._event_postcheck(event)
                        if acceptable:
                            if event.type == "FINISHED":
                                context = f"{self.name}.finish()"
                                try:
                                    await self.run_task(self.finish(), context)
                                except asyncio.CancelledError:
                                    self.debug(f"{context} was cancelled")
                                    continue
                            else:
                                context = f"{self.name}.handle_event({event})"
                                self.scan.stats.event_consumed(event, self)
                                self.debug(f"Handling {event}")