diff --git a/README.md b/README.md index 558d4fb9c9..1f6bbc4e1f 100644 --- a/README.md +++ b/README.md @@ -397,6 +397,7 @@ For details, see [Configuration](https://www.blacklanternsecurity.com/bbot/Stabl - [Setting Up a Dev Environment](https://www.blacklanternsecurity.com/bbot/Stable/dev/dev_environment) - [BBOT Internal Architecture](https://www.blacklanternsecurity.com/bbot/Stable/dev/architecture) - [How to Write a BBOT Module](https://www.blacklanternsecurity.com/bbot/Stable/dev/module_howto) + - [Validating & Inspecting Presets](https://www.blacklanternsecurity.com/bbot/Stable/dev/preset_validation) - [Unit Tests](https://www.blacklanternsecurity.com/bbot/Stable/dev/tests) - [Discord Bot Example](https://www.blacklanternsecurity.com/bbot/Stable/dev/discord_bot) - **Code Reference** diff --git a/bbot/cli.py b/bbot/cli.py index f27bd48e7c..2ebe9046e3 100755 --- a/bbot/cli.py +++ b/bbot/cli.py @@ -102,6 +102,7 @@ async def _main(): preset._default_internal_modules = [] # Bake a temporary copy of the preset so that flags correctly enable their associated modules before listing them + preset.validate() preset = preset.bake() # --list-modules @@ -156,6 +157,7 @@ async def _main(): print(row) return + preset.validate() baked_preset = preset.bake() # --current-preset / --current-preset-full diff --git a/bbot/core/config/files.py b/bbot/core/config/files.py index 2be7bbaa1a..7814626f11 100644 --- a/bbot/core/config/files.py +++ b/bbot/core/config/files.py @@ -1,7 +1,8 @@ import sys +import yaml from pathlib import Path -from omegaconf import OmegaConf +from .merge import deep_merge from ...logger import log_to_stderr from ...errors import ConfigLoadError @@ -18,24 +19,32 @@ class BBOTConfigFiles: def __init__(self, core): self.core = core - def _get_config(self, filename, name="config"): + def _get_config(self, filename, name="config") -> dict: filename = Path(filename).resolve() + if not filename.exists(): + return {} try: - conf = OmegaConf.load(str(filename)) + with open(filename) as f: + conf = yaml.safe_load(f) or {} + if not isinstance(conf, dict): + raise ConfigLoadError( + f"Error parsing config at {filename}: expected a YAML mapping at the top level, " + f"got {type(conf).__name__}" + ) cli_silent = any(x in sys.argv for x in ("-s", "--silent")) if __name__ == "__main__" and not cli_silent: log_to_stderr(f"Loaded {name} from {filename}") return conf + except ConfigLoadError: + raise except Exception as e: - if filename.exists(): - raise ConfigLoadError(f"Error parsing config at {filename}:\n\n{e}") - return OmegaConf.create() + raise ConfigLoadError(f"Error parsing config at {filename}:\n\n{e}") - def get_custom_config(self): - return OmegaConf.merge( + def get_custom_config(self) -> dict: + return deep_merge( self._get_config(self.config_filename, name="config"), self._get_config(self.secrets_filename, name="secrets"), ) - def get_default_config(self): + def get_default_config(self) -> dict: return self._get_config(self.defaults_filename, name="defaults") diff --git a/bbot/core/config/merge.py b/bbot/core/config/merge.py new file mode 100644 index 0000000000..a59964995e --- /dev/null +++ b/bbot/core/config/merge.py @@ -0,0 +1,94 @@ +""" +Deep-merge helpers replacing omegaconf's merge semantics. + +`deep_merge(a, b)` returns a new dict that is `a` with `b` merged in: nested +dicts are merged recursively, leaf values (and lists) from `b` replace those in +`a`. This matches `OmegaConf.merge(a, b)` for BBOT's preset layering use case. +""" + +from __future__ import annotations + +from copy import deepcopy +from typing import Any + + +def deep_merge(base: dict[str, Any] | None, *updates: dict[str, Any] | None) -> dict[str, Any]: + """ + Deep-merge one or more update dicts into a copy of `base`. Last wins on + leaf conflicts; lists are replaced wholesale (not concatenated). + + The returned dict shares no mutable state with the inputs — nested dicts, + lists, and other mutable values are deep-copied as they're carried over. + """ + result: dict[str, Any] = deepcopy(base) if base else {} + for update in updates: + if not update: + continue + for k, v in update.items(): + if k in result and isinstance(result[k], dict) and isinstance(v, dict): + result[k] = deep_merge(result[k], v) + else: + result[k] = deepcopy(v) + return result + + +def dotted_get(data: dict[str, Any], path: str, default: Any = None) -> Any: + """ + Look up a dotted path in a nested dict. + + Note: keys containing literal `.` are not addressable (no escape syntax). + + >>> dotted_get({"a": {"b": {"c": 1}}}, "a.b.c") + 1 + >>> dotted_get({"a": 1}, "a.b.c", default="x") + 'x' + """ + cursor: Any = data + for part in path.split("."): + if not isinstance(cursor, dict) or part not in cursor: + return default + cursor = cursor[part] + return cursor + + +def dotted_set(data: dict[str, Any], path: str, value: Any) -> None: + """ + Set a dotted path in a nested dict, creating intermediate dicts as needed. + + Non-dict intermediates are silently replaced. This is intentional — + callers (CLI parsing) feed the result through pydantic validation, which + surfaces any resulting type mismatch. + + >>> d = {} + >>> dotted_set(d, "a.b.c", 1) + >>> d + {'a': {'b': {'c': 1}}} + """ + parts = path.split(".") + cursor = data + for part in parts[:-1]: + if part not in cursor or not isinstance(cursor[part], dict): + cursor[part] = {} + cursor = cursor[part] + cursor[parts[-1]] = value + + +def iter_dotted_paths(data: dict[str, Any], prefix: str = "") -> list[str]: + """ + Return every dotted leaf path in a nested dict. Empty dicts are treated + as leaves (so they round-trip through dotted_get/dotted_set). + + >>> iter_dotted_paths({"a": 1, "b": {"c": 2}}) + ['a', 'b.c'] + """ + paths: list[str] = [] + for k, v in data.items(): + path = f"{prefix}.{k}" if prefix else k + if isinstance(v, dict) and v: + paths.extend(iter_dotted_paths(v, path)) + else: + paths.append(path) + return paths + + +__all__ = ["deep_merge", "dotted_get", "dotted_set", "iter_dotted_paths"] diff --git a/bbot/core/config/models.py b/bbot/core/config/models.py new file mode 100644 index 0000000000..91cb188cdc --- /dev/null +++ b/bbot/core/config/models.py @@ -0,0 +1,485 @@ +""" +Pydantic schema for BBOT's global config and preset files. + +These models describe the *shape* of valid BBOT configuration — field names +and their expected types — so that `validate_preset()` can catch typos +(`scpoe:`, `http_timoeut:`) and type errors at the boundary. + +Defaults live in `bbot/defaults.yml` — the single source of truth. This file +intentionally does **not** repeat those values; every field is optional, and +an absent field passes validation. At runtime, `BBOTConfigFiles` loads the +merged dict straight from YAML, and these models only ever validate shape. +""" + +from __future__ import annotations + +import os +from typing import Annotated, Any, Literal, Optional + +from pydantic import BaseModel, BeforeValidator, ConfigDict, field_validator +from pydantic import Field as _PydanticField +from pydantic_core import PydanticUndefined + +from bbot.core.helpers.validators import validate_fqdn_or_ip + + +STRICT = ConfigDict(extra="forbid") + + +def _normalize_upper(v): + """Uppercase a string so severity/confidence options are case-insensitive.""" + return v.upper() if isinstance(v, str) else v + + +# Single source of truth for severity/confidence option types (used by the baddns family). +# The BeforeValidator normalizes case at validation time so e.g. "low" validates as "LOW". +SeverityLiteral = Annotated[Literal["INFO", "LOW", "MEDIUM", "HIGH", "CRITICAL"], BeforeValidator(_normalize_upper)] +ConfidenceLiteral = Annotated[ + Literal["UNKNOWN", "LOW", "MEDIUM", "HIGH", "CONFIRMED"], BeforeValidator(_normalize_upper) +] + + +def Field(default=PydanticUndefined, *, sensitive: bool = False, mandatory: bool = False, **kwargs): + """ + Drop-in replacement for `pydantic.Field` that records two BBOT-specific + flags as field metadata: + + - `sensitive=True`: value should be redacted when serializing configs + (api keys, passwords, http cookies, …). + - `mandatory=True`: option must be supplied for the module to function; + drives the "Needs API Key" column in `bbot -l` and the + `BaseModule.auth_required` property. + + Both flags are stashed under `json_schema_extra` so pydantic preserves + them on `FieldInfo.json_schema_extra` (and in any generated JSON schema) + without affecting validation. All other arguments pass through unchanged. + """ + extra = dict(kwargs.pop("json_schema_extra", None) or {}) + if sensitive: + extra["sensitive"] = True + if mandatory: + extra["mandatory"] = True + if extra: + kwargs["json_schema_extra"] = extra + return _PydanticField(default, **kwargs) + + +def field_flags(field) -> dict: + """Return the BBOT flags dict for a pydantic FieldInfo (empty if none).""" + extra = getattr(field, "json_schema_extra", None) + return dict(extra) if isinstance(extra, dict) else {} + + +def is_sensitive(field) -> bool: + return bool(field_flags(field).get("sensitive")) + + +def is_mandatory(field) -> bool: + return bool(field_flags(field).get("mandatory")) + + +def _unwrap_optional(annotation): + """Strip a single `Optional[X]` / `Union[X, None]` wrapper, return X. Pass-through otherwise.""" + import typing + + origin = typing.get_origin(annotation) + if origin is typing.Union: + args = [a for a in typing.get_args(annotation) if a is not type(None)] + if len(args) == 1: + return args[0] + return annotation + + +def _resolve_field(model, key): + """Resolve `key` against `model.model_fields`, honoring `Field(alias=...)`.""" + fields = getattr(model, "model_fields", None) + if not fields: + return None + if key in fields: + return fields[key] + for f in fields.values(): + if getattr(f, "alias", None) == key: + return f + return None + + +def _field_submodel(field): + """If `field`'s annotation is a `BaseModel` subclass, return it; else None.""" + if field is None: + return None + ann = _unwrap_optional(field.annotation) + if isinstance(ann, type) and issubclass(ann, BaseModel): + return ann + return None + + +def _yaml_scalar(value): + """yaml.safe_load a raw string, but never coerce to date/time. Non-strings pass through.""" + import datetime as _dt + import yaml as _yaml + + if not isinstance(value, str): + return value + if value == "": + return "" + try: + parsed = _yaml.safe_load(value) + except _yaml.YAMLError: + return value + return value if isinstance(parsed, (_dt.date, _dt.time)) else parsed + + +def coerce_value(value, adapter): + """Coerce one config value toward its declared type via a pydantic TypeAdapter. + + `value` may be a raw CLI string OR an already-parsed YAML value. `adapter` is + the field's TypeAdapter (from the config type index), or None when the field is + unknown -- then fall back to YAML scalar parsing; unknown keys are still caught + by validation. Returns `value` unchanged if it can't be validated as the declared + type, so the schema pass reports the real error instead of coercion hiding it. + """ + from pydantic import ValidationError + + # pydantic won't coerce os.PathLike -> str, so do it up front. + if isinstance(value, os.PathLike): + value = str(value) + + if adapter is None: + return _yaml_scalar(value) if isinstance(value, str) else value + + if isinstance(value, str): + parsed = _yaml_scalar(value) + # Keep the raw string for scalar fields (lossless: "1.10", "0755", dates, + # bad YAML); only the parsed form can satisfy a list/dict-typed field. + primary = parsed if isinstance(parsed, (list, dict)) else value + fallback = parsed if primary is value else value + for candidate in (primary, fallback): + try: + return adapter.validate_python(candidate) + except ValidationError: + pass + return value + + try: + return adapter.validate_python(value) + except ValidationError: + return value + + +def coerce_config(config, index, prefix=""): + """Walk a config dict and coerce each leaf toward its declared type.""" + out = {} + for k, v in config.items(): + dotted = f"{prefix}.{k}" if prefix else k + if isinstance(v, dict): + out[k] = coerce_config(v, index, dotted) + else: + out[k] = coerce_value(v, index.get(dotted)) + return out + + +def partition_sensitive_config(config, model, *, keep_sensitive: bool): + """ + Walk `config` (a dict) alongside the pydantic `model`, returning a copy + that either drops or extracts every `sensitive=True` field. + + - `keep_sensitive=False` -> caller wants the public, non-secret view (used + by `BBOTCore.no_secrets_config`). + - `keep_sensitive=True` -> caller wants only the secrets (used by + `BBOTCore.secrets_only_config` to materialize `~/.config/bbot/secrets.yml`). + + Unknown keys (no matching field in the schema) pass through unchanged when + redacting and are dropped when extracting secrets-only. + """ + import copy as _copy + + if not isinstance(config, dict) or model is None: + return _copy.deepcopy(config) if keep_sensitive is False else {} + + out: dict = {} + for key, val in config.items(): + field = _resolve_field(model, key) + sub_model = _field_submodel(field) + + if sub_model is not None and isinstance(val, dict): + child = partition_sensitive_config(val, sub_model, keep_sensitive=keep_sensitive) + # When redacting, preserve the parent key even if its body was + # entirely sensitive — matches the prior `clean_dict` behavior. + # When extracting secrets-only, drop empty branches so the result + # is just the secrets that exist. + if keep_sensitive: + if child: + out[key] = child + else: + out[key] = child + continue + + if field is None: + # No matching schema field — pass through unchanged when redacting, + # drop when extracting secrets-only. + if not keep_sensitive: + out[key] = _copy.deepcopy(val) + continue + + sensitive = is_sensitive(field) + if keep_sensitive: + if sensitive: + out[key] = _copy.deepcopy(val) + else: + if not sensitive: + out[key] = _copy.deepcopy(val) + return out + + +class ScopeConfig(BaseModel): + model_config = STRICT + + strict: Optional[bool] = None + report_distance: Optional[int] = None + search_distance: Optional[int] = None + + +class DnsConfig(BaseModel): + model_config = STRICT + + disable: Optional[bool] = None + minimal: Optional[bool] = None + threads: Optional[int] = None + cache_size: Optional[int] = None + brute_threads: Optional[int] = None + brute_nameservers: Optional[str] = None + search_distance: Optional[int] = None + runaway_limit: Optional[int] = None + timeout: Optional[int] = None + retries: Optional[int] = None + wildcard_disable: Optional[bool] = None + wildcard_ignore: Optional[list[str]] = None + wildcard_tests: Optional[int] = None + abort_threshold: Optional[int] = None + filter_ptrs: Optional[bool] = None + debug: Optional[bool] = None + omit_queries: Optional[list[str]] = None + + +class BodySpillConfig(BaseModel): + """`web.body_spill` — keeps large HTTP_RESPONSE bodies off the Python heap.""" + + model_config = STRICT + + enabled: Optional[bool] = None + cache_mb: Optional[int] = None + compress: Optional[bool] = None + + +class WebConfig(BaseModel): + model_config = STRICT + + http_proxy: Optional[str] = None + http_proxy_exclude: Optional[list[str]] = None + user_agent: Optional[str] = None + user_agent_suffix: Optional[str] = None + spider_distance: Optional[int] = None + spider_depth: Optional[int] = None + spider_links_per_page: Optional[int] = None + http_timeout: Optional[int] = None + blasthttp_timeout: Optional[int] = None + http_headers: Optional[dict[str, str]] = None + http_cookies: Optional[dict[str, str]] = Field(default=None, sensitive=True) + api_retries: Optional[int] = None + http_retries: Optional[int] = None + blasthttp_retries: Optional[int] = None + http_rate_limit: Optional[int] = None + body_spill: Optional[BodySpillConfig] = None + # The `429_*` keys start with a digit, so we expose them via aliases. + sleep_interval_429: Optional[int] = Field(default=None, alias="429_sleep_interval") + max_sleep_interval_429: Optional[int] = Field(default=None, alias="429_max_sleep_interval") + debug: Optional[bool] = None + http_max_redirects: Optional[int] = None + ssl_verify: Optional[bool] = None + + +class EngineConfig(BaseModel): + model_config = STRICT + + debug: Optional[bool] = None + + +class DepsToolConfig(BaseModel): + """Per-tool dep config (e.g. `deps.ffuf.version`).""" + + model_config = STRICT + + version: Optional[str] = None + + +class DepsConfig(BaseModel): + model_config = STRICT + + behavior: Optional[Literal["abort_on_failure", "retry_failed", "ignore_failed", "disable", "force_install"]] = None + ffuf: Optional[DepsToolConfig] = None + + +class BaseModuleConfig(BaseModel): + """ + Shared base for every module's `class Config(BaseModuleConfig)`. + + Declares the three universal module options that are applied to every + module regardless of declaration. The actual default values live in + `bbot/defaults.yml`; this class only validates shape. + """ + + model_config = STRICT + + batch_size: Optional[int] = Field( + default=None, + description="The number of events to process in a single batch (only applies to batch modules)", + ) + module_threads: Optional[int] = Field( + default=None, + description="How many event handlers to run in parallel", + ) + module_timeout: Optional[int] = Field( + default=None, + description="Max time in seconds to spend handling each event or batch of events", + ) + + +class BBOTConfig(BaseModel): + """ + Root BBOT config schema. Unknown top-level keys are rejected so that + typos like `scpoe:` or `moudules:` become loud errors instead of silent + no-ops. + + This is a validation schema only -- it has no default values. The real + defaults live in `bbot/defaults.yml`. + """ + + model_config = ConfigDict( + extra="forbid", + populate_by_name=True, + ) + + # Basic options + home: Optional[str] = None + keep_scans: Optional[int] = None + status_frequency: Optional[int] = None + file_blobs: Optional[bool] = None + folder_blobs: Optional[bool] = None + max_mem_percent: Optional[int] = None + + # Nested sections + scope: Optional[ScopeConfig] = None + dns: Optional[DnsConfig] = None + web: Optional[WebConfig] = None + engine: Optional[EngineConfig] = None + deps: Optional[DepsConfig] = None + + # Module loader paths + module_dirs: Optional[list[str]] = None + + # Module runtime + module_handle_event_timeout: Optional[int] = None + module_handle_batch_timeout: Optional[int] = None + + # Internal module toggles (hardcoded because they're first-class scan + # pipeline features; the set changes rarely) + speculate: Optional[bool] = None + excavate: Optional[bool] = None + aggregate: Optional[bool] = None + dnsresolve: Optional[bool] = None + cloudcheck: Optional[bool] = None + unarchive: Optional[bool] = None + + # URL handling + url_querystring_remove: Optional[bool] = None + url_querystring_collapse: Optional[bool] = None + url_extension_blacklist: Optional[list[str]] = None + url_extension_special: Optional[list[str]] = None + url_extension_static: Optional[list[str]] = None + + # Parameter handling + parameter_blacklist: Optional[list[str]] = None + parameter_blacklist_prefixes: Optional[list[str]] = None + + # Event output filter + omit_event_types: Optional[list[str]] = None + + # Interactsh + interactsh_server: Optional[str] = None + interactsh_token: Optional[str] = Field(default=None, sensitive=True) + interactsh_disable: Optional[bool] = None + + # bbot.io API key (used by the asn helper) + bbot_io_api_key: Optional[str] = Field(default=None, sensitive=True) + + _validate_interactsh_server = field_validator("interactsh_server")(validate_fqdn_or_ip) + + # Per-module configs — validated separately, per-module, against each + # module's own `class Config(BaseModuleConfig)`. + modules: Optional[dict[str, dict[str, Any]]] = None + + +class PresetSchema(BaseModel): + """ + Schema for the top-level keys in a preset YAML file. Catches typos like + `modlues:` or `flgas:` at load time. + + `target`/`targets` and `include`/`presets` are aliases; both are accepted. + The `config` key is validated separately as `BBOTConfig`. + """ + + model_config = ConfigDict( + extra="forbid", + populate_by_name=True, + ) + + target: Optional[list[str]] = None + targets: Optional[list[str]] = None + seeds: Optional[list[str]] = None + blacklist: Optional[list[str]] = None + + modules: Optional[list[str]] = None + output_modules: Optional[list[str]] = None + exclude_modules: Optional[list[str]] = None + flags: Optional[list[str]] = None + require_flags: Optional[list[str]] = None + exclude_flags: Optional[list[str]] = None + + config: Optional[dict[str, Any]] = None + module_dirs: Optional[list[str]] = None + + include: Optional[list[str]] = None + presets: Optional[list[str]] = None + + scan_name: Optional[str] = None + output_dir: Optional[str] = None + name: Optional[str] = None + description: Optional[str] = None + + conditions: Optional[list[str]] = None + + verbose: Optional[bool] = None + debug: Optional[bool] = None + silent: Optional[bool] = None + + +__all__ = [ + "BBOTConfig", + "BaseModuleConfig", + "ConfidenceLiteral", + "DepsConfig", + "DepsToolConfig", + "DnsConfig", + "EngineConfig", + "Field", + "PresetSchema", + "ScopeConfig", + "SeverityLiteral", + "WebConfig", + "coerce_config", + "coerce_value", + "field_flags", + "is_mandatory", + "is_sensitive", + "partition_sensitive_config", +] diff --git a/bbot/core/core.py b/bbot/core/core.py index 5814052771..feeccac0f5 100644 --- a/bbot/core/core.py +++ b/bbot/core/core.py @@ -1,15 +1,14 @@ import os import logging -from copy import copy +from copy import copy, deepcopy from pathlib import Path -from contextlib import suppress -from omegaconf import OmegaConf from bbot.errors import BBOTError +from .config.merge import deep_merge from .multiprocess import SHARED_INTERPRETER_STATE -DEFAULT_CONFIG = None +DEFAULT_CONFIG: dict | None = None class BBOTCore: @@ -26,17 +25,12 @@ class BBOTCore: - load quickly """ - # used for filtering out sensitive config values - secrets_strings = ["api_key", "username", "password", "token", "secret", "_id"] - # don't filter/remove entries under this key - secrets_exclude_keys = ["modules"] - def __init__(self): self._logger = None self._files_config = None - self._config = None - self._custom_config = None + self._config: dict | None = None + self._custom_config: dict | None = None # bare minimum == logging self.logger @@ -85,22 +79,20 @@ def scans_dir(self): return self.home / "scans" @property - def config(self): + def config(self) -> dict: """ - .config is just .default_config + .custom_config merged together + .config is just .default_config + .custom_config merged together. - any new values should be added to custom_config. + Any new values should be added to custom_config. """ if self._config is None: - self._config = OmegaConf.merge(self.default_config, self.custom_config) - # set read-only flag (change .custom_config instead) - OmegaConf.set_readonly(self._config, True) + self._config = deep_merge(self.default_config, self.custom_config) return self._config @property - def default_config(self): + def default_config(self) -> dict: """ - The default BBOT config (from `defaults.yml`). Read-only. + The default BBOT config (from `defaults.yml`). """ global DEFAULT_CONFIG if DEFAULT_CONFIG is None: @@ -111,16 +103,14 @@ def default_config(self): return DEFAULT_CONFIG @default_config.setter - def default_config(self, value): + def default_config(self, value: dict): # we temporarily clear out the config so it can be refreshed if/when default_config changes global DEFAULT_CONFIG self._config = None - DEFAULT_CONFIG = value - # set read-only flag (change .custom_config instead) - OmegaConf.set_readonly(DEFAULT_CONFIG, True) + DEFAULT_CONFIG = dict(value) if value else {} @property - def custom_config(self): + def custom_config(self) -> dict: """ Custom BBOT config (from `~/.config/bbot/bbot.yml`) """ @@ -131,59 +121,59 @@ def custom_config(self): return self._custom_config @custom_config.setter - def custom_config(self, value): - # we temporarily clear out the config so it can be refreshed if/when custom_config changes + def custom_config(self, value: dict): self._config = None - # ensure the modules key is always a dictionary - modules_entry = value.get("modules", None) - if modules_entry is not None and not OmegaConf.is_dict(modules_entry): - value["modules"] = {} - self._custom_config = value + self._custom_config = dict(value) if value else {} def no_secrets_config(self, config): - from .helpers.misc import clean_dict + """Return a copy of `config` with every `sensitive=True` field removed. - with suppress(ValueError): - config = OmegaConf.to_object(config) + Sensitivity is read from the per-field `json_schema_extra["sensitive"]` + flag declared on `BBOTConfig` (and each module's `class Config`). + Module-level redaction uses the composite schema built lazily by + `MODULE_LOADER.config_schema`; if a key isn't covered by any schema + (e.g. an unknown module), it passes through unchanged. + """ + from .config.models import partition_sensitive_config - return clean_dict( - config, - *self.secrets_strings, - fuzzy=True, - exclude_keys=self.secrets_exclude_keys, - ) + return partition_sensitive_config(config, self._config_schema(), keep_sensitive=False) def secrets_only_config(self, config): - from .helpers.misc import filter_dict + """Return a copy of `config` containing only `sensitive=True` fields. + + Inverse of `no_secrets_config()`. Useful for splitting a merged config + into a public `bbot.yml` and a private `secrets.yml`. + """ + from .config.models import partition_sensitive_config - with suppress(ValueError): - config = OmegaConf.to_object(config) + return partition_sensitive_config(config, self._config_schema(), keep_sensitive=True) - return filter_dict( - config, - *self.secrets_strings, - fuzzy=True, - exclude_keys=self.secrets_exclude_keys, - ) + def _config_schema(self): + """Resolve the runtime BBOTConfig schema (with per-module configs).""" + try: + from bbot.core.modules import MODULE_LOADER + + return MODULE_LOADER.config_schema + except Exception: + from .config.models import BBOTConfig + + return BBOTConfig def merge_custom(self, config): - """ - Merge a config into the custom config. - """ - self.custom_config = OmegaConf.merge(self.custom_config, OmegaConf.create(config)) + """Merge a config dict into the custom config.""" + self.custom_config = deep_merge(self.custom_config, dict(config) if config else {}) def merge_default(self, config): - """ - Merge a config into the default config. - """ - self.default_config = OmegaConf.merge(self.default_config, OmegaConf.create(config)) + """Merge a config dict into the default config.""" + self.default_config = deep_merge(self.default_config, dict(config) if config else {}) def copy(self): """ Return a semi-shallow copy of self. (`custom_config` is copied, but `default_config` stays the same) """ core_copy = copy(self) - core_copy._custom_config = self._custom_config.copy() + core_copy._custom_config = deepcopy(self._custom_config) if self._custom_config else {} + core_copy._config = None return core_copy @property diff --git a/bbot/core/helpers/misc.py b/bbot/core/helpers/misc.py index da34e051fc..8ae4238ce7 100644 --- a/bbot/core/helpers/misc.py +++ b/bbot/core/helpers/misc.py @@ -2818,36 +2818,23 @@ def truncate_filename(file_path, max_length=255): def get_keys_in_dot_syntax(config): - """Retrieve all keys in an OmegaConf configuration in dot notation. - - This function converts an OmegaConf configuration into a list of keys - represented in dot notation. + """Retrieve all leaf keys in a nested dict in dot notation. Args: - config (DictConfig): The OmegaConf configuration object. + config (dict): A nested dict. Returns: - List[str]: A list of keys in dot notation. + List[str]: A list of leaf keys in dot notation. Examples: - >>> config = OmegaConf.create({ - ... "web": { - ... "test": True - ... }, - ... "db": { - ... "host": "localhost", - ... "port": 5432 - ... } - ... }) - >>> get_keys_in_dot_syntax(config) + >>> get_keys_in_dot_syntax({"web": {"test": True}, "db": {"host": "localhost", "port": 5432}}) ['web.test', 'db.host', 'db.port'] """ - from omegaconf import OmegaConf - - container = OmegaConf.to_container(config, resolve=True) keys = [] def recursive_keys(d, parent_key=""): + if not isinstance(d, dict): + return for k, v in d.items(): full_key = f"{parent_key}.{k}" if parent_key else k if isinstance(v, dict): @@ -2855,7 +2842,7 @@ def recursive_keys(d, parent_key=""): else: keys.append(full_key) - recursive_keys(container) + recursive_keys(config) return keys diff --git a/bbot/core/helpers/names_generator.py b/bbot/core/helpers/names_generator.py index 31501aec72..13d43c007b 100644 --- a/bbot/core/helpers/names_generator.py +++ b/bbot/core/helpers/names_generator.py @@ -237,6 +237,7 @@ "rapid_unscheduled", "raving", "reckless", + "recursive", "reductive", "ripped", "ruthless", @@ -720,6 +721,7 @@ "theoden", "theon", "theresa", + "thetechromancer", "thomas", "tiffany", "timothy", diff --git a/bbot/core/helpers/validators.py b/bbot/core/helpers/validators.py index 27c8bd0fea..577f124756 100644 --- a/bbot/core/helpers/validators.py +++ b/bbot/core/helpers/validators.py @@ -7,7 +7,7 @@ from bbot.core.helpers import regexes from bbot.errors import ValidationError from bbot.core.helpers.url import parse_url, hash_url -from bbot.core.helpers.misc import smart_encode_punycode, split_host_port, make_netloc, is_ip +from bbot.core.helpers.misc import smart_encode_punycode, split_host_port, make_netloc, is_dns_name, is_ip log = logging.getLogger("bbot.core.helpers.validators") @@ -129,6 +129,38 @@ def validate_host(host: Union[str, ipaddress.IPv4Address, ipaddress.IPv6Address] raise ValidationError(f'Invalid hostname: "{host}"') +def validate_fqdn_or_ip(host): + """Strict FQDN-or-IP check. Accepts an IPv4/IPv6 address or a hostname + containing at least one dot. Rejects single-label values (e.g. + `localhost`, or a domain typed without its TLD), so domain-shaped + config fields fail at preset-load time instead of surfacing as runtime + errors deep in a scan. + + `None` and the empty string pass through to support optional config + fields with `null` defaults. + + Unlike `validate_host`, this function does not normalize the input + (no port-stripping, lowercasing, or punycode conversion); it returns + the value as supplied. Use it as a pydantic `field_validator` on + schema fields that must be an FQDN or IP and nothing else. + + Examples: + >>> validate_fqdn_or_ip("example.com") + 'example.com' + >>> validate_fqdn_or_ip("192.168.1.1") + '192.168.1.1' + >>> validate_fqdn_or_ip("localhost") + ValueError: not a valid FQDN or IP address: 'localhost' + """ + if host is None or host == "": + return host + if is_ip(host): + return host + if isinstance(host, str) and "." in host and is_dns_name(host): + return host + raise ValueError(f"not a valid FQDN or IP address: {host!r}") + + FINDING_SEVERITY_LEVELS = ("INFO", "LOW", "MEDIUM", "HIGH", "CRITICAL") diff --git a/bbot/core/modules.py b/bbot/core/modules.py index f89f114fe7..f0ec78aa04 100644 --- a/bbot/core/modules.py +++ b/bbot/core/modules.py @@ -1,15 +1,14 @@ import re import ast import sys +import yaml import atexit import pickle import logging import importlib -import omegaconf import traceback from copy import copy from pathlib import Path -from omegaconf import OmegaConf from contextlib import suppress from bbot.core import CORE @@ -34,6 +33,229 @@ 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 = 3 + + +_UNEVALUATED = object() + + +def _eval_ast_default(node): + """ + Extract a literal default value from an AST node. Returns _UNEVALUATED if + the node can't be resolved statically (e.g. `default_factory=lambda: ...` + or a non-constant expression). Preload uses this for display-time defaults + only; actual validation happens at bake time against the real pydantic + Config class. + """ + if node is None: + return _UNEVALUATED + try: + return ast.literal_eval(node) + except (ValueError, TypeError, SyntaxError): + # Recognize a few common default_factory values. + if isinstance(node, ast.Name): + return {"list": [], "dict": {}, "set": set(), "tuple": (), "str": "", "int": 0, "float": 0.0}.get( + node.id, _UNEVALUATED + ) + return _UNEVALUATED + + +def _exec_config_class(source: str, module_name: str): + """ + Execute a `class Config(BaseModuleConfig):` snippet in a controlled + namespace and return the resulting class. The namespace provides exactly + what a Config block is allowed to reference: the typing primitives, + pydantic's `Field` factory and validator decorators, and `BaseModuleConfig`. + + This replaces parsing annotations as strings: pydantic handles every + valid type expression (`Optional[str]`, `Literal["a", "b"]`, + `list[Union[int, str]]`, …) without any hand-rolled resolver. + """ + from typing import Annotated, Any, Dict, List, Literal, Optional, Set, Tuple, Union + from pydantic import AfterValidator, BeforeValidator, field_validator, model_validator + from bbot.core.config.models import BaseModuleConfig, ConfidenceLiteral, Field, SeverityLiteral + + namespace: dict = { + "Annotated": Annotated, + "Any": Any, + "Dict": Dict, + "List": List, + "Literal": Literal, + "Optional": Optional, + "Set": Set, + "Tuple": Tuple, + "Union": Union, + "Field": Field, + "field_validator": field_validator, + "model_validator": model_validator, + "BeforeValidator": BeforeValidator, + "AfterValidator": AfterValidator, + "BaseModuleConfig": BaseModuleConfig, + "SeverityLiteral": SeverityLiteral, + "ConfidenceLiteral": ConfidenceLiteral, + } + try: + exec(source, namespace) + except Exception as e: + raise BBOTError( + f'module "{module_name}" has an invalid Config class ({type(e).__name__}: {e}). ' + "Config blocks may only reference: Optional, Union, Literal, Any, List, Dict, Tuple, Set, " + "Field, field_validator, model_validator, BeforeValidator, AfterValidator, BaseModuleConfig, " + "and Python builtins." + ) from e + cfg = namespace.get("Config") + if cfg is None: + raise BBOTError(f'module "{module_name}": Config snippet did not define a class named "Config"') + return cfg + + +def _build_validation_schema(preloaded: dict): + """ + Build the composite preset validation schema. + + Structure: + FullPresetSchema + ├── (all PresetSchema fields — target, modules, flags, …) + └── config: FullBBOTConfig + ├── (all BBOTConfig fields — scope, dns, web, …) + └── modules: ModulesSchema + ├── nuclei: NucleiModuleConfig + ├── httpx: HttpxModuleConfig + ├── sslcert: SslcertModuleConfig + └── … one field per known module + + A single `FullPresetSchema.model_validate(preset_dict)` call then catches + every class of error in one pass: + - top-level preset typos (extra='forbid' on PresetSchema) + - global config typos / wrong types (extra='forbid' on BBOTConfig) + - unknown module names (extra='forbid' on ModulesSchema) + - wrong module option names / wrong types (extra='forbid' per module) + """ + import warnings + from typing import Optional + from pydantic import ConfigDict, Field, create_model + from bbot.core.config.models import BaseModuleConfig, BBOTConfig, PresetSchema + + module_fields = {} + for name, data in preloaded.items(): + source = data.get("config_source") + if not source: + # Module declares no Config — only the universal options apply. + field_type = BaseModuleConfig + else: + try: + field_type = _exec_config_class(source, name) + except BBOTError as e: + # The Config references a name not available in the isolated exec + # namespace (a module-level constant, imported type, Enum, …); it's + # valid at real import time. Don't reject the module — accept any + # config for it (its own options just aren't strictly validated here). + log.debug(f"{e} -- accepting any config for module '{name}'") + field_type = dict + module_fields[name] = (Optional[field_type], Field(default=None)) + + # Some module names (e.g. `json`) shadow BaseModel's deprecated method + # names and trigger a UserWarning. The field still validates correctly; + # silence the noise. + with warnings.catch_warnings(): + warnings.filterwarnings( + "ignore", + message=r'Field name ".+" in ".+" shadows an attribute in parent "BaseModel"', + category=UserWarning, + ) + ModulesSchema = create_model( + "ModulesSchema", + __config__=ConfigDict(extra="forbid"), + **module_fields, + ) + FullBBOTConfig = create_model( + "FullBBOTConfig", + __base__=BBOTConfig, + modules=(Optional[ModulesSchema], Field(default=None)), + ) + FullPresetSchema = create_model( + "FullPresetSchema", + __base__=PresetSchema, + config=(Optional[FullBBOTConfig], Field(default=None)), + ) + return FullPresetSchema + + +def _extract_pydantic_config(config_class: ast.ClassDef) -> tuple[dict, dict, set, set, dict]: + """ + Walk a `class Config(BaseModuleConfig):` block and extract + `(defaults, descriptions, sensitive, mandatory, types)` -- cheap metadata + used for `bbot -l`, type-directed config coercion, and similar listing + paths, without importing or exec-ing anything. + + `types` maps each field name to its raw annotation string (e.g. + `"bool"`, `"Union[str, list[str]]"`, `"Literal['manual', 'severe']"`). + The actual typed pydantic class is built later via `_exec_config_class` + on the captured source text. + """ + defaults: dict = {} + descriptions: dict = {} + sensitive: set = set() + mandatory: set = set() + types: dict = {} + for node in config_class.body: + # `model_config = ConfigDict(...)` etc. are plain assigns, not typed -- skip. + if not isinstance(node, ast.AnnAssign) or not isinstance(node.target, ast.Name): + continue + name = node.target.id + if name.startswith("_"): + continue + + types[name] = ast.unparse(node.annotation) + default = _UNEVALUATED + description = "" + is_sensitive = False + is_mandatory = False + + value = node.value + if isinstance(value, ast.Call) and isinstance(value.func, ast.Name) and value.func.id == "Field": + # Field(default, description="...", default_factory=..., sensitive=..., mandatory=..., ...) + # - first positional arg is the default, if given + if value.args: + default = _eval_ast_default(value.args[0]) + for kw in value.keywords: + if kw.arg == "default": + default = _eval_ast_default(kw.value) + elif kw.arg == "default_factory": + default = _eval_ast_default(kw.value) + elif kw.arg == "description": + with suppress(ValueError, TypeError, SyntaxError): + description = ast.literal_eval(kw.value) + elif kw.arg == "sensitive": + with suppress(ValueError, TypeError, SyntaxError): + is_sensitive = bool(ast.literal_eval(kw.value)) + elif kw.arg == "mandatory": + with suppress(ValueError, TypeError, SyntaxError): + is_mandatory = bool(ast.literal_eval(kw.value)) + elif kw.arg == "json_schema_extra": + with suppress(ValueError, TypeError, SyntaxError): + extra = ast.literal_eval(kw.value) + if isinstance(extra, dict): + is_sensitive = is_sensitive or bool(extra.get("sensitive")) + is_mandatory = is_mandatory or bool(extra.get("mandatory")) + elif value is not None: + default = _eval_ast_default(value) + + if default is _UNEVALUATED: + # couldn't statically determine; fall back to None so listing still works + default = None + defaults[name] = default + descriptions[name] = description + if is_sensitive: + sensitive.add(name) + if is_mandatory: + mandatory.add(name) + return defaults, descriptions, sensitive, mandatory, types + + class ModuleLoader: """ Main class responsible for preloading BBOT modules. @@ -64,6 +286,10 @@ def __init__(self): self.internal_module_choices = set() self._preload_cache = None + # Composite preset-validation schema, built from preloaded modules. + # Invalidated whenever a new module is preloaded. + self._validation_schema = None + self._config_type_index = None self._module_dirs = set() self._module_dirs_preloaded = set() @@ -145,7 +371,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: @@ -178,6 +404,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") @@ -196,18 +428,18 @@ def preload(self, module_dirs=None): self.flag_choices.update(set(flags)) self.__preloaded[module_name] = preloaded - config = OmegaConf.create(preloaded.get("config", {})) - self._configs[module_name] = config + self._configs[module_name] = dict(preloaded.get("config", {})) self._module_dirs_preloaded.add(module_dir) # update default config with module defaults - module_config = omegaconf.OmegaConf.create( - { - "modules": self.configs(), - } - ) - self.core.merge_default(module_config) + self.core.merge_default({"modules": self.configs()}) + + # invalidate the composite validation schema; it'll rebuild lazily + # on next access now that the set of modules has changed + if new_modules: + self._validation_schema = None + self._config_type_index = None return new_modules @@ -254,12 +486,94 @@ def preloaded(self, type=None): return preloaded def configs(self, type=None): - configs = {} if type is not None: - configs = {k: v for k, v in self._configs.items() if self.check_type(k, type)} - else: - configs = dict(self._configs) - return OmegaConf.create(configs) + return {k: dict(v) for k, v in self._configs.items() if self.check_type(k, type)} + return {k: dict(v) for k, v in self._configs.items()} + + @property + def validation_schema(self): + """ + The composite pydantic schema for validating a full preset dict. + + Built lazily from preloaded module metadata and cached. Rebuilt when + `preload()` discovers new modules (e.g. after `add_module_dir`). + """ + if self._validation_schema is None: + self._validation_schema = _build_validation_schema(self._preloaded) + return self._validation_schema + + @property + def config_schema(self): + """ + The runtime `BBOTConfig` schema with per-module configs grafted in. + + This is `validation_schema.config` (i.e. `FullBBOTConfig`) and is the + right model to walk a config dict against — used by + `BBOTCore.no_secrets_config()` and `BBOTCore.secrets_only_config()` to + partition sensitive fields. + """ + from bbot.core.config.models import _unwrap_optional + + field = self.validation_schema.model_fields.get("config") + if field is None: + from bbot.core.config.models import BBOTConfig + + return BBOTConfig + return _unwrap_optional(field.annotation) + + @property + def config_type_index(self): + """{dotted_config_path: TypeAdapter} for every known config option. + + Built by walking the materialized config schema (`config_schema`, i.e. + global config + every module's exec'd Config), so a single pydantic + TypeAdapter per leaf field drives coercion -- no hand-rolled type-name + parsing. Adapters are memoized by annotation (many fields share e.g. + `Optional[str]`). Nested models are recursed into via `_field_submodel`, + so `modules..