Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 43 additions & 1 deletion bbot/core/modules.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,12 @@
bbot_code_dir = Path(__file__).parent.parent


# Bump when the preloader's output schema or validation rules change. Folded
# into the per-module cache_key so stale entries from older bbot versions get
# rebuilt instead of silently bypassing new checks.
PRELOAD_CACHE_VERSION = 2


_UNEVALUATED = object()


Expand Down Expand Up @@ -349,7 +355,7 @@ def preload(self, module_dirs=None):
module_file = module_file.resolve()

# try to load from cache
module_cache_key = (str(module_file), tuple(module_file.stat()))
module_cache_key = (PRELOAD_CACHE_VERSION, str(module_file), tuple(module_file.stat()))
preloaded = self.preload_cache.get(module_name, {})
cache_key = preloaded.get("cache_key", ())
if preloaded and module_cache_key == cache_key:
Expand Down Expand Up @@ -382,6 +388,12 @@ def preload(self, module_dirs=None):
preloaded["namespace"] = namespace
preloaded["cache_key"] = module_cache_key

except BBOTError as e:
# Intentional, user-facing errors raised from preload_module
# (e.g. the pre-3.0 options-dict migration message). Skip the
# traceback so the message reads cleanly.
log_to_stderr(str(e), level="CRITICAL")
sys.exit(1)
except Exception:
log_to_stderr(f"Error preloading {module_file}\n\n{traceback.format_exc()}", level="CRITICAL")
log_to_stderr(f"Error in {module_file.name}", level="CRITICAL")
Expand Down Expand Up @@ -563,6 +575,8 @@ def preload_module(self, module_file):
options_sensitive: set = set()
options_mandatory: set = set()
config_source: str | None = None
legacy_options_keyword: str | None = None
legacy_options_line: int | None = None
disable_auto_module_deps = False
with open(module_file) as f:
python_code = f.read()
Expand Down Expand Up @@ -603,6 +617,15 @@ def preload_module(self, module_file):
if not type(class_attr) == ast.Assign:
continue

# legacy options/options_desc dict: record so we can fail
# with a migration hint after the walk completes
if legacy_options_keyword is None:
for target in class_attr.targets:
if isinstance(target, ast.Name) and target.id in ("options", "options_desc"):
legacy_options_keyword = target.id
legacy_options_line = class_attr.lineno
break

# module metadata
if type(class_attr.value) == ast.Dict:
if any(target.id == "meta" for target in class_attr.targets):
Expand Down Expand Up @@ -664,6 +687,25 @@ def preload_module(self, module_file):
if type(class_attr.value.value) == bool:
disable_auto_module_deps = class_attr.value.value

# Reject the pre-3.0 options/options_desc dict format. Those dicts are
# silently ignored by the new loader: defaults disappear and setting
# them via -c or YAML emits misleading "did you mean ...?" suggestions
# pointing at unrelated modules.
module_name = module_file.stem
if legacy_options_keyword is not None and config_source is None:
raise BBOTError(
f'Module "{module_name}" ({module_file}:{legacy_options_line}) declares a legacy '
f"`{legacy_options_keyword}` dict, which is no longer supported in BBOT 3.0+.\n\n"
f"Replace the `options` / `options_desc` dicts with a nested\n"
f"`class Config(BaseModuleConfig)` schema. For example:\n\n"
f" from bbot.core.config.models import BaseModuleConfig, Field\n\n"
f" class {module_name}(BaseModule):\n"
f" ...\n"
f" class Config(BaseModuleConfig):\n"
f' api_key: str = Field("", description="API key", sensitive=True)\n'
f' max_results: int = Field(100, description="Max results to return")\n'
)

for task in ansible_tasks:
if "become" not in task:
task["become"] = False
Expand Down
83 changes: 83 additions & 0 deletions bbot/test/test_step_1/test_modules_basic.py
Original file line number Diff line number Diff line change
Expand Up @@ -541,3 +541,86 @@ async def test_module_loading(bbot_scanner):
assert not any(not_async)

await scan2._cleanup()


def test_legacy_options_dict_rejected(tmp_path):
"""A pre-3.0 user module that declares an `options` / `options_desc` dict
instead of a `class Config(BaseModuleConfig)` schema must be rejected at
preload time with a clear migration message. Otherwise the loader would
silently ignore the defaults and the user would never know."""
from bbot.errors import BBOTError
from bbot.core.modules import MODULE_LOADER

legacy_mod = tmp_path / "legacy_user_mod.py"
legacy_mod.write_text(
"""
from bbot.modules.base import BaseModule


class legacy_user_mod(BaseModule):
watched_events = ["DNS_NAME"]
produced_events = ["FINDING"]
flags = ["passive", "safe"]
meta = {"description": "x", "created_date": "2024-01-01", "author": "@x"}

options = {"api_key": "", "max_results": 100}
options_desc = {"api_key": "API key", "max_results": "Max results"}

async def handle_event(self, event):
pass
"""
)

with pytest.raises(BBOTError) as excinfo:
MODULE_LOADER.preload_module(legacy_mod)
msg = str(excinfo.value)
assert "legacy_user_mod" in msg
assert "no longer supported" in msg
assert "class Config(BaseModuleConfig)" in msg

# And a module that declares only options_desc (still legacy) is also rejected.
legacy_mod_desc_only = tmp_path / "legacy_desc_only.py"
legacy_mod_desc_only.write_text(
"""
from bbot.modules.base import BaseModule


class legacy_desc_only(BaseModule):
watched_events = ["DNS_NAME"]
produced_events = ["FINDING"]
flags = ["passive", "safe"]
meta = {"description": "x", "created_date": "2024-01-01", "author": "@x"}

options_desc = {"api_key": "API key"}

async def handle_event(self, event):
pass
"""
)
with pytest.raises(BBOTError):
MODULE_LOADER.preload_module(legacy_mod_desc_only)

# A module with a real `class Config` and NO legacy dict must load cleanly.
new_mod = tmp_path / "new_style_mod.py"
new_mod.write_text(
"""
from bbot.modules.base import BaseModule
from bbot.core.config.models import BaseModuleConfig, Field


class new_style_mod(BaseModule):
watched_events = ["DNS_NAME"]
produced_events = ["FINDING"]
flags = ["passive", "safe"]
meta = {"description": "x", "created_date": "2024-01-01", "author": "@x"}

class Config(BaseModuleConfig):
api_key: str = Field("", description="API key", sensitive=True)

async def handle_event(self, event):
pass
"""
)
preloaded = MODULE_LOADER.preload_module(new_mod)
assert preloaded is not None
assert preloaded["config"] == {"api_key": ""}
Loading