Method _worker

bbot/modules/base.py:1805–1876

(self)

    _intercept = True

    async def _worker(self):
        async with self.scan._acatch(context=self._worker, unhandled_is_critical=True):
            try:
                while not self.scan.stopping and not self.errored:
                    try:
                        if self.incoming_event_queue is not False:
                            incoming = await self.get_incoming_event()
                            try:
                                event, kwargs = incoming
                            except ValueError:
                                event = incoming
                                kwargs = {}
                        else:
                            self.debug("Event queue is in bad state")
                            break
                    except asyncio.queues.QueueEmpty:
                        await asyncio.sleep(0.1)
                        continue

                    if event.type == "FINISHED":
                        context = f"{self.name}.finish()"
                        async with self.scan._acatch(context), self._task_counter.count(context):
                            await self.finish()
                        continue

                    acceptable = True
                    async with self._task_counter.count(f"event_precheck({event})"):
                        precheck_pass, reason = self._event_precheck(event)
                    if not precheck_pass:
                        self.debug(f"Not intercepting {event} because precheck failed ({reason})")
                        acceptable = False
                    async with self._task_counter.count(f"event_postcheck({event})"):
                        postcheck_pass, reason = await self._event_postcheck(event)
                    if not postcheck_pass:
                        self.debug(f"Not intercepting {event} because postcheck failed ({reason})")
                        acceptable = False

                    # whether to pass the event on to the rest of the scan
                    # defaults to true, unless handle_event returns False
                    forward_event = True
                    forward_event_reason = ""

                    if acceptable:
                        context = f"{self.name}.handle_event({event, kwargs})"
                        self.scan.stats.event_consumed(event, self)
                        self.debug(f"Intercepting {event}")
                        try:
                            forward_event = await self.run_task(self.handle_event(event, **kwargs), context)
                        except asyncio.CancelledError:
                            self.debug(f"{context} was cancelled")
                            continue
                        with suppress(ValueError, TypeError):
                            forward_event, forward_event_reason = forward_event

                        if forward_event is False:
                            self.debug(f"Not forwarding {event} because {forward_event_reason}")
                            continue