Main class responsible for preloading BBOT modules. This class is in charge of preloading modules to determine their dependencies. Once dependencies are identified, they are installed before the actual module is imported. This ensures that all requisite libraries and components are
| 41 | |
| 42 | |
| 43 | class ModuleLoader: |
| 44 | """ |
| 45 | Main class responsible for preloading BBOT modules. |
| 46 | |
| 47 | This class is in charge of preloading modules to determine their dependencies. |
| 48 | Once dependencies are identified, they are installed before the actual module is imported. |
| 49 | This ensures that all requisite libraries and components are available for the module to function correctly. |
| 50 | """ |
| 51 | |
| 52 | default_module_dir = bbot_code_dir / "modules" |
| 53 | |
| 54 | module_dir_regex = re.compile(r"^[a-z][a-z0-9_]*$") |
| 55 | |
| 56 | # if a module consumes these event types, automatically assume these dependencies |
| 57 | default_module_deps = {"HTTP_RESPONSE": "httpx", "URL": "httpx", "SOCIAL": "social"} |
| 58 | |
| 59 | def __init__(self): |
| 60 | self.core = CORE |
| 61 | |
| 62 | self._shared_deps = dict(SHARED_DEPS) |
| 63 | |
| 64 | self.__preloaded = {} |
| 65 | self._configs = {} |
| 66 | self.flag_choices = set() |
| 67 | self.all_module_choices = set() |
| 68 | self.scan_module_choices = set() |
| 69 | self.output_module_choices = set() |
| 70 | self.internal_module_choices = set() |
| 71 | |
| 72 | self._preload_cache = None |
| 73 | |
| 74 | self._module_dirs = set() |
| 75 | self._module_dirs_preloaded = set() |
| 76 | self.add_module_dir(self.default_module_dir) |
| 77 | |
| 78 | # save preload cache before exiting |
| 79 | atexit.register(self.save_preload_cache) |
| 80 | |
| 81 | def copy(self): |
| 82 | module_loader_copy = copy(self) |
| 83 | module_loader_copy.__preloaded = dict(self.__preloaded) |
| 84 | return module_loader_copy |
| 85 | |
| 86 | @property |
| 87 | def preload_cache_file(self): |
| 88 | return self.core.cache_dir / "module_preload_cache" |
| 89 | |
| 90 | @property |
| 91 | def module_dirs(self): |
| 92 | return self._module_dirs |
| 93 | |
| 94 | def add_module_dir(self, module_dir): |
| 95 | module_dir = Path(module_dir).resolve() |
| 96 | if module_dir in self._module_dirs: |
| 97 | log.debug(f'Already added custom module dir "{module_dir}"') |
| 98 | return |
| 99 | if not module_dir.is_dir(): |
| 100 | log.warning(f'Failed to add custom module dir "{module_dir}", please make sure it exists') |
no test coverage detected
searching dependent graphs…