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
149 changes: 33 additions & 116 deletions bbot/core/config/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -113,87 +113,6 @@ def _field_submodel(field):
return None


_COLLECTION_NAMES = frozenset({"list", "List", "tuple", "Tuple", "set", "Set", "frozenset", "dict", "Dict"})
_TRUE_WORDS = frozenset({"true", "1", "yes", "on"})
_FALSE_WORDS = frozenset({"false", "0", "no", "off"})


def accepted_types_from_string(src: str) -> frozenset:
"""Base type-names a stringified annotation accepts (no exec, pure AST).

Used to normalize module Config annotations captured at preload time into
a set of base type-names that `coerce_value` can act on.

>>> accepted_types_from_string("bool")
frozenset({'bool'})
>>> accepted_types_from_string("Optional[int]")
frozenset({'int'})
>>> accepted_types_from_string("Union[str, list[str]]")
frozenset({'str', 'list'})
>>> accepted_types_from_string('Literal["manual", "x"]')
frozenset({'str'})
"""
import ast as _ast

try:
root = _ast.parse(src, mode="eval").body
except SyntaxError:
return frozenset()

def walk(n):
t: set = set()
if isinstance(n, _ast.Name):
t.add(n.id)
elif isinstance(n, _ast.Attribute):
t.add(n.attr)
elif isinstance(n, _ast.BinOp) and isinstance(n.op, _ast.BitOr):
t |= walk(n.left) | walk(n.right)
elif isinstance(n, _ast.Subscript):
base = getattr(n.value, "id", getattr(n.value, "attr", None))
elts = n.slice.elts if isinstance(n.slice, _ast.Tuple) else [n.slice]
if base in ("Optional", "Union"):
for e in elts:
t |= walk(e)
elif base == "Literal":
t |= {type(e.value).__name__ for e in elts if isinstance(e, _ast.Constant)}
elif base == "Annotated":
if elts:
t |= walk(elts[0])
else:
t.add(base)
return t

return frozenset(x for x in walk(root) if x != "NoneType")


def accepted_types_from_annotation(annotation) -> frozenset:
"""Base type-names a real annotation object accepts.

Used for the static global models (BBOTConfig, WebConfig, etc.) where the
annotation is a live type object, not a string.
"""
import typing

origin = typing.get_origin(annotation)
if origin is typing.Literal:
return frozenset(type(a).__name__ for a in typing.get_args(annotation))
args = typing.get_args(annotation) if origin is typing.Union else (annotation,)
out: set = set()
for a in args:
if a is type(None):
continue
ao = typing.get_origin(a)
if ao is typing.Literal:
out |= {type(x).__name__ for x in typing.get_args(a)}
elif ao is typing.Annotated or (hasattr(typing, "Annotated") and typing.get_origin(a) is typing.Annotated):
inner_args = typing.get_args(a)
if inner_args:
out |= accepted_types_from_annotation(inner_args[0])
else:
out.add(getattr(ao or a, "__name__", str(ao or a)))
return frozenset(out)


def _yaml_scalar(value):
"""yaml.safe_load a raw string, but never coerce to date/time. Non-strings pass through."""
import datetime as _dt
Expand All @@ -210,41 +129,41 @@ def _yaml_scalar(value):
return value if isinstance(parsed, (_dt.date, _dt.time)) else parsed


def coerce_value(value, accepted):
"""Coerce one config value toward its declared type.
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.
`accepted` is the frozenset of base type-names for the target field, or
None/empty when the field is unknown (fall back to default YAML behavior;
unknown keys are still caught by validation).
`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.
"""
is_raw = isinstance(value, str)
if not accepted:
return _yaml_scalar(value) if is_raw else value
if "str" in accepted and not (accepted & _COLLECTION_NAMES):
if is_raw:
return value
if value is None or isinstance(value, (list, dict, set)):
return value
return str(value)
if accepted == frozenset({"bool"}):
v = _yaml_scalar(value) if is_raw else value
if isinstance(v, bool):
return v
if isinstance(v, (int, float)):
return bool(v)
if isinstance(v, str):
low = v.strip().lower()
if low in _TRUE_WORDS:
return True
if low in _FALSE_WORDS:
return False
return v
if is_raw:
return _yaml_scalar(value)
if "str" in accepted and isinstance(value, os.PathLike):
return str(value)
return value
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=""):
Expand Down Expand Up @@ -557,8 +476,6 @@ class PresetSchema(BaseModel):
"ScopeConfig",
"SeverityLiteral",
"WebConfig",
"accepted_types_from_annotation",
"accepted_types_from_string",
"coerce_config",
"coerce_value",
"field_flags",
Expand Down
64 changes: 39 additions & 25 deletions bbot/core/modules.py
Original file line number Diff line number Diff line change
Expand Up @@ -523,41 +523,55 @@ def config_schema(self):

@property
def config_type_index(self):
"""{dotted_config_path: frozenset(base_type_names)} for every known option.

Global keys come from the static BBOTConfig tree; per-module keys from
the preloaded AST annotation strings. No module Config is exec'd.
"""{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.<name>.<option>` keys fall out of the same walk.
"""
if self._config_type_index is None:
from bbot.core.config.models import (
BBOTConfig,
BaseModuleConfig,
accepted_types_from_annotation,
accepted_types_from_string,
_field_submodel,
)
from pydantic import ConfigDict, TypeAdapter
from bbot.core.config.models import _field_submodel

# coerce_numbers_to_str lets a numeric YAML value (e.g. password: 12345678)
# validate against a str field; pydantic is otherwise lax-but-not-that-lax.
adapter_config = ConfigDict(coerce_numbers_to_str=True)
adapter_cache = {}

def make_adapter(annotation):
try:
if annotation in adapter_cache:
return adapter_cache[annotation]
hashable = True
except TypeError:
hashable = False
try:
adapter = TypeAdapter(annotation, config=adapter_config)
except Exception as e:
log.debug(f"config_type_index: no TypeAdapter for {annotation!r}: {e}")
adapter = None
if hashable:
adapter_cache[annotation] = adapter
return adapter

index = {}

def walk_static(model, prefix=""):
def walk(model, prefix=""):
for fname, field in model.model_fields.items():
sub = _field_submodel(field)
for key in {fname, getattr(field, "alias", None)} - {None}:
dotted = f"{prefix}.{key}" if prefix else key
sub = _field_submodel(field)
if sub is not None and fname != "modules":
walk_static(sub, dotted)
if sub is not None:
walk(sub, dotted)
else:
index[dotted] = accepted_types_from_annotation(field.annotation)

walk_static(BBOTConfig)

universal = {n: f.annotation for n, f in BaseModuleConfig.model_fields.items()}
for name, data in self._preloaded.items():
for field, ann_str in data.get("options_types", {}).items():
index[f"modules.{name}.{field}"] = accepted_types_from_string(ann_str)
for field, ann in universal.items():
index.setdefault(f"modules.{name}.{field}", accepted_types_from_annotation(ann))
adapter = make_adapter(field.annotation)
if adapter is not None:
index[dotted] = adapter

walk(self.config_schema)
self._config_type_index = index
return self._config_type_index

Expand Down
18 changes: 9 additions & 9 deletions bbot/scanner/preset/args.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,18 +8,18 @@
from bbot.core.helpers.misc import chain_lists


def _parse_cli_value(raw: str, accepted=None):
def _parse_cli_value(raw: str, adapter=None):
"""Parse the RHS of a `-c a.b.c=value` argument.

`accepted` is the frozenset of base type-names for the target field (from the
config type index), or None when the field is unknown. Coercion follows the
declared type: string fields keep the literal text (lossless), bool fields
produce a real bool, int/float fields use YAML, and unknown fields fall back to
plain YAML coercion.
`adapter` is the target field's pydantic TypeAdapter (from the config type
index), or None when the field is unknown. Coercion follows the declared type:
string fields keep the literal text (lossless), bool fields produce a real bool,
int/float fields parse via YAML, and unknown fields fall back to plain YAML
coercion.
"""
if raw == "":
return ""
return coerce_value(raw, accepted)
return coerce_value(raw, adapter)


def parse_dotted_cli(entries, index=None):
Expand All @@ -37,8 +37,8 @@ def parse_dotted_cli(entries, index=None):
path = path.strip()
if not path:
raise ValueError(f'Empty key in "{entry}"')
accepted = index.get(path) if index is not None else None
dotted_set(result, path, _parse_cli_value(raw.strip(), accepted))
adapter = index.get(path) if index is not None else None
dotted_set(result, path, _parse_cli_value(raw.strip(), adapter))
return result


Expand Down
Loading