Pre-checks an event to determine if it should be accepted by the module for queuing. This method is called when an event is about to be enqueued into the module's incoming event queue. It applies various filters such as special signal event types, module error state, watche
(self, event)
| 804 | return max(0, self.scan.scope_search_distance + self.scope_distance_modifier) |
| 805 | |
| 806 | def _event_precheck(self, event): |
| 807 | """ |
| 808 | Pre-checks an event to determine if it should be accepted by the module for queuing. |
| 809 | |
| 810 | This method is called when an event is about to be enqueued into the module's incoming event queue. |
| 811 | It applies various filters such as special signal event types, module error state, watched event types, and more |
| 812 | to decide whether or not the event should be enqueued. |
| 813 | |
| 814 | Args: |
| 815 | event (Event): The event object to check. |
| 816 | |
| 817 | Returns: |
| 818 | tuple: A tuple (bool, str) where the bool indicates if the event should be accepted, and the str gives the reason. |
| 819 | |
| 820 | Examples: |
| 821 | >>> result, reason = self._event_precheck(event) |
| 822 | >>> if result: |
| 823 | ... self.incoming_event_queue.put_nowait(event) |
| 824 | ... else: |
| 825 | ... self.debug(f"Not accepting {event} because {reason}") |
| 826 | |
| 827 | Notes: |
| 828 | - The method considers special signal event types like "FINISHED". |
| 829 | - Checks whether the module is in an error state. |
| 830 | - Checks if the event type matches the types this module is interested in (`watched_events`). |
| 831 | - Checks for events tagged as 'target' if the module has `target_only` flag set. |
| 832 | - Applies specific filtering based on event type and module name. |
| 833 | """ |
| 834 | |
| 835 | # special signal event types |
| 836 | if event.type in ("FINISHED",): |
| 837 | return True, "its type is FINISHED" |
| 838 | if self.errored: |
| 839 | return False, "module is in error state" |
| 840 | # exclude non-watched types |
| 841 | if not any(t in self.get_watched_events() for t in ("*", event.type)): |
| 842 | return False, "its type is not in watched_events" |
| 843 | if self.target_only: |
| 844 | if "target" not in event.tags: |
| 845 | return False, "it did not meet target_only filter criteria" |
| 846 | |
| 847 | # limit js URLs to modules that opt in to receive them |
| 848 | if (not self.accept_url_special) and event.type.startswith("URL"): |
| 849 | extension = getattr(event, "url_extension", "") |
| 850 | if extension in self.scan.url_extension_special: |
| 851 | return ( |
| 852 | False, |
| 853 | f"it is a special URL (extension {extension}) but the module does not opt in to receive special URLs", |
| 854 | ) |
| 855 | |
| 856 | return True, "precheck succeeded" |
| 857 | |
| 858 | async def _event_postcheck(self, event): |
| 859 | """ |