mirror/nanobot
1
0
Fork 0
mirror of https://github.com/HKUDS/nanobot.git synced 2026-08-04 16:38:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(config): centralize file persistence

Browse Source
This commit is contained in:
chengyongru 2026-07-14 11:28:48 +08:00
parent 6e462e62a6
commit f19efcd990
16 changed files with 1149 additions and 655 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
.agent/design.md
View File

@ -27,3 +27,9 @@ A bugfix should make the protected invariant clear, change the smallest surface
## Explicit over magical
Configuration must be declared explicitly in `config/schema.py` Pydantic models. Error handling should raise clear exceptions rather than silently correcting bad input. Provider auto-detection exists, but every resolution path must be traceable from the factory to the concrete provider class.
## Configuration has an explicit owner
`FileConfigRepository` owns config-file reads, validation, revisions, and atomic writes. Runtime code may use the compatibility helpers in `config/loader.py`, but new components with an explicit config path should keep their own repository instance instead of importing a mutable global config object.
Persisted and runtime views are separate: `load_raw()` preserves `${VAR}` references for editing, while `load_effective()` returns an isolated snapshot with references resolved. Read-modify-write flows must use `update()` / `update_config()` so a stale object cannot silently overwrite a newer change. Loading config is side-effect free; process-wide policies are applied explicitly during runtime startup.

2
.agent/gotchas.md
View File

@ -6,7 +6,7 @@
## Config `${VAR}` References
`config/loader.py` resolves `${VAR}` patterns in `config.json` at load time. This is **not** a shell-like default-value syntax. If the environment variable is missing, `load_config` raises `ValueError` and the agent falls back to default configuration.
`load_config()` and `FileConfigRepository.load_raw()` preserve `${VAR}` patterns so Settings can safely edit and save the persisted representation. Runtime entry points use `load_effective_config()` / `load_effective()` to resolve them in an isolated snapshot. This is **not** a shell-like default-value syntax. If a referenced variable is missing, effective loading raises `ValueError`.
Example valid usage:
```json

2
.agent/security.md
View File

@ -16,7 +16,7 @@ Shell execution (`ExecTool`, `agent/tools/shell.py`) also respects `restrict_to_
All outbound HTTP requests from agent tools must pass through `validate_url_target` (`security/network.py`). By default it blocks loopback, RFC1918 private addresses, CGNAT ranges, link-local ranges, and cloud metadata endpoints (including `169.254.169.254`).
The only escape hatch is `configure_ssrf_whitelist(cidrs)`, which reads from `config.tools.ssrf_whitelist` at load time.
The only escape hatch is `configure_ssrf_whitelist(cidrs)`. Runtime entry points explicitly apply `config.tools.ssrf_whitelist` after loading the effective config; ordinary config reads must not mutate this process-wide policy.
HTTP/SSE MCP transports are part of this boundary: validate configured MCP URLs before probing or constructing clients, and validate each outgoing HTTP request before redirects are followed. Local/private HTTP MCP endpoints are allowed only through the explicit SSRF whitelist. Stdio MCP servers are not part of the HTTP SSRF path.

161
nanobot/channels/feishu.py
View File

@ -626,38 +626,44 @@ def sync_saved_feishu_identity_boundary(
if not current_identity_key:
return False
from nanobot.config.loader import load_config, save_config
full_config = load_config()
feishu_cfg = getattr(full_config.channels, "feishu", None) or {}
if not isinstance(feishu_cfg, dict):
feishu_cfg = {}
from nanobot.config.loader import update_config
defaults = FeishuChannel.default_config()
previous_identity_key = ""
for spec in feishu_instance_specs(feishu_cfg, defaults):
if spec.instance_id == instance_id:
previous_identity_key = str(
spec.config.get("identityKey") or spec.config.get("identity_key") or ""
)
break
access_cleared = False
access_cleared = bool(previous_identity_key and previous_identity_key != current_identity_key)
values: dict[str, Any] = {"identityKey": current_identity_key}
if access_cleared:
values["allowFrom"] = []
values["allow_from"] = []
clear_channel(runtime_channel_name("feishu", instance_id))
def mutate(full_config: Any) -> None:
nonlocal access_cleared
feishu_cfg = getattr(full_config.channels, "feishu", None) or {}
if not isinstance(feishu_cfg, dict):
feishu_cfg = {}
if not previous_identity_key or access_cleared:
feishu_cfg = update_feishu_instance_preserving_shape(
feishu_cfg,
defaults,
instance_id,
values,
previous_identity_key = ""
for spec in feishu_instance_specs(feishu_cfg, defaults):
if spec.instance_id == instance_id:
previous_identity_key = str(
spec.config.get("identityKey") or spec.config.get("identity_key") or ""
)
break
access_cleared = bool(
previous_identity_key and previous_identity_key != current_identity_key
)
setattr(full_config.channels, "feishu", feishu_cfg)
save_config(full_config)
values: dict[str, Any] = {"identityKey": current_identity_key}
if access_cleared:
values["allowFrom"] = []
values["allow_from"] = []
if not previous_identity_key or access_cleared:
full_config.channels.feishu = update_feishu_instance_preserving_shape(
feishu_cfg,
defaults,
instance_id,
values,
)
update_config(mutate)
if access_cleared:
clear_channel(runtime_channel_name("feishu", instance_id))
return access_cleared
@ -669,19 +675,13 @@ def save_registration_result(
name: str | None = None,
) -> None:
"""Persist a successful Feishu/Lark registration result to config.json."""
from nanobot.config.loader import load_config, save_config
from nanobot.config.loader import update_config
full_config = load_config()
feishu_cfg = getattr(full_config.channels, "feishu", None) or {}
if not isinstance(feishu_cfg, dict):
feishu_cfg = {}
defaults = FeishuChannel.default_config()
app_id = str(result["app_id"]).strip()
domain = str(result.get("domain", "feishu") or "feishu").strip().lower()
domain = "lark" if domain == "lark" else "feishu"
previous_identity_key = _saved_feishu_instance_identity_key(feishu_cfg, defaults, instance_id)
next_identity_key = _feishu_app_identity_key(app_id, domain)
identity_changed = bool(previous_identity_key and previous_identity_key != next_identity_key)
identity: dict[str, str] = {}
with suppress(Exception):
identity = fetch_feishu_app_identity(
@ -698,18 +698,35 @@ def save_registration_result(
"enabled": True,
**identity,
}
identity_changed = False
def mutate(full_config: Any) -> None:
nonlocal identity_changed
feishu_cfg = getattr(full_config.channels, "feishu", None) or {}
if not isinstance(feishu_cfg, dict):
feishu_cfg = {}
previous_identity_key = _saved_feishu_instance_identity_key(
feishu_cfg,
defaults,
instance_id,
)
identity_changed = bool(
previous_identity_key and previous_identity_key != next_identity_key
)
next_values = dict(values)
if identity_changed:
next_values["allowFrom"] = []
next_values["allow_from"] = []
full_config.channels.feishu = upsert_feishu_instance(
feishu_cfg,
defaults,
instance_id,
next_values,
)
update_config(mutate)
if identity_changed:
values["allowFrom"] = []
values["allow_from"] = []
clear_channel(runtime_channel_name("feishu", instance_id))
feishu_cfg = upsert_feishu_instance(
feishu_cfg,
defaults,
instance_id,
values,
)
setattr(full_config.channels, "feishu", feishu_cfg)
save_config(full_config)
def refresh_saved_feishu_identities(config: Any | None = None) -> bool:
@ -723,13 +740,13 @@ def refresh_saved_feishu_identities(config: Any | None = None) -> bool:
if not FEISHU_AVAILABLE:
return False
from nanobot.config.loader import load_config, save_config
from nanobot.config.loader import load_config, update_config
full_config = config or load_config()
feishu_cfg = getattr(full_config.channels, "feishu", None)
source_config = config or load_config()
feishu_cfg = getattr(source_config.channels, "feishu", None)
defaults = FeishuChannel.default_config()
specs = feishu_instance_specs(feishu_cfg, defaults)
updated = False
fetched: dict[str, tuple[tuple[str, str, str], dict[str, str]]] = {}
for spec in specs:
instance = spec.config
@ -753,20 +770,50 @@ def refresh_saved_feishu_identities(config: Any | None = None) -> bool:
if not identity:
identity = {"identityFetchedAt": _identity_timestamp()}
feishu_cfg = update_feishu_instance_preserving_shape(
feishu_cfg,
defaults,
spec.instance_id,
fetched[spec.instance_id] = (
(app_id, app_secret, str(instance.get("domain") or "feishu")),
identity,
)
updated = True
if not updated:
if not fetched:
return False
setattr(full_config.channels, "feishu", feishu_cfg)
save_config(full_config)
return True
updated = False
def mutate(full_config: Any) -> None:
nonlocal updated
current_cfg = getattr(full_config.channels, "feishu", None)
for spec in feishu_instance_specs(current_cfg, defaults):
fetched_entry = fetched.get(spec.instance_id)
if fetched_entry is None:
continue
expected_credentials, identity = fetched_entry
instance = spec.config
current_credentials = (
str(instance.get("appId") or instance.get("app_id") or "").strip(),
str(instance.get("appSecret") or instance.get("app_secret") or "").strip(),
str(instance.get("domain") or "feishu"),
)
if current_credentials != expected_credentials:
continue
if (
instance.get("displayName")
or instance.get("avatarUrl")
or instance.get("identityFetchedAt")
):
continue
current_cfg = update_feishu_instance_preserving_shape(
current_cfg,
defaults,
spec.instance_id,
identity,
)
updated = True
if updated:
full_config.channels.feishu = current_cfg
update_config(mutate)
return updated
def qr_register(

49
nanobot/cli/commands.py
View File

@ -738,27 +738,25 @@ def onboard(
def _onboard_plugins(config_path: Path) -> None:
"""Inject default config for all discovered channels (built-in + plugins)."""
import json
from nanobot.channels.registry import discover_all
from nanobot.config.loader import merge_missing_defaults
from nanobot.config.loader import get_config_repository, merge_missing_defaults
all_channels = discover_all()
if not all_channels:
return
with open(config_path, encoding="utf-8") as f:
data = json.load(f)
def mutate(config: Config) -> None:
for name, cls in all_channels.items():
existing = getattr(config.channels, name, None)
if not isinstance(existing, dict):
existing = {}
setattr(
config.channels,
name,
merge_missing_defaults(existing, cls.default_config()),
)
channels = data.setdefault("channels", {})
for name, cls in all_channels.items():
if name not in channels:
channels[name] = cls.default_config()
else:
channels[name] = merge_missing_defaults(channels[name], cls.default_config())
with open(config_path, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2, ensure_ascii=False)
get_config_repository(config_path).update(mutate)
def _print_enable_options(
@ -798,7 +796,11 @@ def _model_display(config: Config) -> tuple[str, str]:
def _load_runtime_config(config: str | None = None, workspace: str | None = None) -> Config:
"""Load config and optionally override the active workspace."""
from nanobot.config.loader import load_config, resolve_config_env_vars, set_config_path
from nanobot.config.loader import (
apply_config_runtime_policies,
load_effective_config,
set_config_path,
)
config_path = None
if config:
@ -810,13 +812,14 @@ def _load_runtime_config(config: str | None = None, workspace: str | None = None
console.print(f"[dim]Using config: {config_path}[/dim]")
try:
loaded = resolve_config_env_vars(load_config(config_path))
loaded = load_effective_config(config_path)
except ValueError as e:
console.print(f"[red]Error: {e}[/red]")
raise typer.Exit(1)
_warn_deprecated_config_keys(config_path)
if workspace:
loaded.agents.defaults.workspace = workspace
apply_config_runtime_policies(loaded)
return loaded
@ -2688,19 +2691,21 @@ def _set_oauth_provider_as_main(
config_path: str | None = None,
) -> None:
"""Persist an OAuth provider as the active agent provider."""
from nanobot.config.loader import get_config_path, load_config, save_config, set_config_path
from nanobot.config.loader import get_config_path, set_config_path, update_config
resolved_config_path = Path(config_path).expanduser().resolve() if config_path else None
if resolved_config_path is not None:
set_config_path(resolved_config_path)
console.print(f"[dim]Using config: {resolved_config_path}[/dim]")
config = load_config(resolved_config_path)
selected_model = (model or "").strip() or _OAUTH_PROVIDER_DEFAULT_MODELS[provider_name]
config.agents.defaults.model_preset = None
config.agents.defaults.provider = provider_name
config.agents.defaults.model = selected_model
save_config(config, resolved_config_path)
def mutate(config: Config) -> None:
config.agents.defaults.model_preset = None
config.agents.defaults.provider = provider_name
config.agents.defaults.model = selected_model
update_config(mutate, resolved_config_path)
saved_path = resolved_config_path or get_config_path()
console.print(

25
nanobot/config/__init__.py
View File

@ -1,6 +1,13 @@
"""Configuration module for nanobot."""
from nanobot.config.loader import get_config_path, load_config
from nanobot.config.loader import (
apply_config_runtime_policies,
get_config_path,
get_config_repository,
load_config,
load_effective_config,
update_config,
)
from nanobot.config.paths import (
get_cli_history_path,
get_cron_dir,
@ -13,12 +20,28 @@ from nanobot.config.paths import (
get_workspace_path,
is_default_workspace,
)
from nanobot.config.repository import (
ConfigCommit,
ConfigConflictError,
EffectiveConfigSnapshot,
FileConfigRepository,
PersistedConfigSnapshot,
)
from nanobot.config.schema import Config
__all__ = [
"Config",
"ConfigCommit",
"ConfigConflictError",
"EffectiveConfigSnapshot",
"FileConfigRepository",
"PersistedConfigSnapshot",
"apply_config_runtime_policies",
"load_config",
"load_effective_config",
"update_config",
"get_config_path",
"get_config_repository",
"get_data_dir",
"get_runtime_subdir",
"get_media_dir",

237
nanobot/config/loader.py
View File

@ -1,92 +1,90 @@
"""Configuration loading utilities."""
"""Compatibility helpers for configuration loading and persistence."""
import json
import os
import re
from __future__ import annotations
from collections.abc import Callable
from pathlib import Path
from typing import Any
import pydantic
from loguru import logger
from pydantic import BaseModel
from loguru import logger as logger # compatibility: callers patch loader.logger
from nanobot.config.schema import Config, _resolve_tool_config_refs
from nanobot.config.repository import (
ConfigCommit,
FileConfigRepository,
resolve_config_env_vars,
)
from nanobot.config.repository import (
_migrate_config as _migrate_config,
)
from nanobot.config.repository import (
_resolve_env_vars as _resolve_env_vars,
)
from nanobot.config.schema import Config
# Global variable to store current config path (for multi-instance support)
# Legacy default-instance path. Runtime code should prefer an explicitly scoped
# FileConfigRepository; these helpers remain for CLI and plugin compatibility.
_current_config_path: Path | None = None
_schema_refs_ready = False
def set_config_path(path: Path) -> None:
"""Set the current config path (used to derive data directory)."""
"""Set the default config path used by compatibility helpers."""
global _current_config_path
_current_config_path = path
def get_config_path() -> Path:
"""Get the configuration file path."""
"""Get the default configuration file path."""
if _current_config_path:
return _current_config_path
return Path.home() / ".nanobot" / "config.json"
def get_config_repository(config_path: Path | None = None) -> FileConfigRepository:
"""Return an instance-scoped repository for *config_path*."""
return FileConfigRepository(config_path or get_config_path())
def load_config(config_path: Path | None = None) -> Config:
"""Load raw persisted config without applying process runtime policy."""
return get_config_repository(config_path).load_raw().config
def load_effective_config(config_path: Path | None = None) -> Config:
"""Load a fresh runtime config with environment references resolved.
Route through ``load_config`` so existing embedders that replace the legacy
loader hook keep working during the repository migration.
"""
Load configuration from file or create default.
Args:
config_path: Optional path to config file. Uses default if not provided.
Returns:
Loaded configuration object.
"""
global _schema_refs_ready
if not _schema_refs_ready:
_resolve_tool_config_refs()
_schema_refs_ready = True
path = config_path or get_config_path()
config = Config()
if path.exists():
try:
with open(path, encoding="utf-8") as f:
data = json.load(f)
data = _migrate_config(data)
config = Config.model_validate(data)
except (json.JSONDecodeError, ValueError, pydantic.ValidationError) as e:
raise ValueError(f"Failed to load config from {path}: {e}") from e
_apply_ssrf_whitelist(config)
return config
def _apply_ssrf_whitelist(config: Config) -> None:
"""Apply SSRF whitelist from config to the network security module."""
from nanobot.security.network import configure_ssrf_whitelist
configure_ssrf_whitelist(config.tools.ssrf_whitelist)
return resolve_config_env_vars(load_config(config_path))
def save_config(config: Config, config_path: Path | None = None) -> None:
"""Atomically save a complete config.
New read-modify-write flows should use :func:`update_config` so concurrent
writers for the same file cannot silently overwrite one another.
"""
Save configuration to file.
get_config_repository(config_path).save(config)
Args:
config: Configuration to save.
config_path: Optional path to save to. Uses default if not provided.
"""
path = config_path or get_config_path()
path.parent.mkdir(parents=True, exist_ok=True)
data = config.model_dump(mode="json", by_alias=True)
if config.providers.openai_codex.proxy is not None:
data.setdefault("providers", {})["openaiCodex"] = {
"proxy": config.providers.openai_codex.proxy,
}
def update_config(
mutator: Callable[[Config], None],
config_path: Path | None = None,
*,
expected_revision: str | None = None,
) -> ConfigCommit:
"""Atomically mutate the latest raw config and return the resulting commit."""
return get_config_repository(config_path).update(
mutator,
expected_revision=expected_revision,
)
with open(path, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2, ensure_ascii=False)
def apply_config_runtime_policies(config: Config) -> None:
"""Apply process-level policies when starting or reconfiguring a runtime."""
from nanobot.security.network import configure_ssrf_whitelist
configure_ssrf_whitelist(config.tools.ssrf_whitelist)
def merge_missing_defaults(existing: Any, defaults: Any) -> Any:
@ -103,109 +101,16 @@ def merge_missing_defaults(existing: Any, defaults: Any) -> Any:
return merged
_ENV_REF_PATTERN = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}")
def resolve_config_env_vars(config: Config) -> Config:
"""Return *config* with ``${VAR}`` env-var references resolved.
Walks in place so fields declared with ``exclude=True`` survive;
returns the same instance when no references are present.
Raises ``ValueError`` if a referenced variable is not set.
"""
return _resolve_in_place(config)
def _resolve_in_place(obj: Any) -> Any:
if isinstance(obj, str):
new = _ENV_REF_PATTERN.sub(_env_replace, obj)
return new if new != obj else obj
if isinstance(obj, BaseModel):
updates: dict[str, Any] = {}
for name in type(obj).model_fields:
old = getattr(obj, name)
new = _resolve_in_place(old)
if new is not old:
updates[name] = new
extras = obj.__pydantic_extra__
new_extras: dict[str, Any] | None = None
if extras:
resolved = {k: _resolve_in_place(v) for k, v in extras.items()}
if any(resolved[k] is not extras[k] for k in extras):
new_extras = resolved
if not updates and new_extras is None:
return obj
copy = obj.model_copy(update=updates) if updates else obj.model_copy()
if new_extras is not None:
copy.__pydantic_extra__ = new_extras
return copy
if isinstance(obj, dict):
resolved = {k: _resolve_in_place(v) for k, v in obj.items()}
return resolved if any(resolved[k] is not obj[k] for k in obj) else obj
if isinstance(obj, list):
resolved = [_resolve_in_place(v) for v in obj]
return resolved if any(nv is not ov for nv, ov in zip(resolved, obj)) else obj
return obj
def _resolve_env_vars(obj: object) -> object:
"""Recursively resolve ``${VAR}`` patterns in plain strings/dicts/lists."""
if isinstance(obj, str):
return _ENV_REF_PATTERN.sub(_env_replace, obj)
if isinstance(obj, dict):
return {k: _resolve_env_vars(v) for k, v in obj.items()}
if isinstance(obj, list):
return [_resolve_env_vars(v) for v in obj]
return obj
def _env_replace(match: re.Match[str]) -> str:
name = match.group(1)
value = os.environ.get(name)
if value is None:
raise ValueError(
f"Environment variable '{name}' referenced in config is not set"
)
return value
def _migrate_config(data: dict) -> dict:
"""Migrate old config formats to current."""
agents = data.get("agents", {})
defaults = agents.get("defaults", {}) if isinstance(agents, dict) else {}
if isinstance(defaults, dict):
had_legacy_max_messages = (
"maxMessages" in defaults or "max_messages" in defaults
)
defaults.pop("maxMessages", None)
defaults.pop("max_messages", None)
if had_legacy_max_messages:
# TODO(next version): Remove this legacy cleanup branch; the schema
# will silently ignore this field once the warning grace period ends.
logger.warning(
"agents.defaults.maxMessages/max_messages is legacy and ignored; "
"replay max messages is now an internal safety cap. Remove it from "
"config. This compatibility warning will be removed in the next version."
)
# Move tools.exec.restrictToWorkspace → tools.restrictToWorkspace
tools = data.get("tools", {})
exec_cfg = tools.get("exec", {})
if "restrictToWorkspace" in exec_cfg and "restrictToWorkspace" not in tools:
tools["restrictToWorkspace"] = exec_cfg.pop("restrictToWorkspace")
# Move tools.myEnabled / tools.mySet → tools.my.{enable, allowSet}.
# The old flat keys shipped in the initial MyTool landing; wrapping them in a
# sub-config keeps `web` / `exec` / `my` symmetric and gives room to grow.
if "myEnabled" in tools or "mySet" in tools:
my_cfg = tools.setdefault("my", {})
if "myEnabled" in tools and "enable" not in my_cfg:
my_cfg["enable"] = tools.pop("myEnabled")
else:
tools.pop("myEnabled", None)
if "mySet" in tools and "allowSet" not in my_cfg:
my_cfg["allowSet"] = tools.pop("mySet")
else:
tools.pop("mySet", None)
return data
__all__ = [
"FileConfigRepository",
"apply_config_runtime_policies",
"get_config_path",
"get_config_repository",
"load_config",
"load_effective_config",
"merge_missing_defaults",
"resolve_config_env_vars",
"save_config",
"set_config_path",
"update_config",
]

319
nanobot/config/repository.py Normal file
View File

@ -0,0 +1,319 @@
"""Instance-scoped configuration persistence and resolution."""
from __future__ import annotations
import hashlib
import json
import os
import re
import stat
import threading
import uuid
from collections.abc import Callable
from contextlib import suppress
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from loguru import logger
from pydantic import BaseModel
from nanobot.config.schema import Config, _resolve_tool_config_refs
ConfigMutator = Callable[[Config], None]
_ENV_REF_PATTERN = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}")
_schema_refs_ready = False
_schema_refs_lock = threading.Lock()
_path_locks_guard = threading.Lock()
_path_locks: dict[Path, threading.RLock] = {}
class ConfigConflictError(RuntimeError):
"""Raised when a caller tries to update a stale configuration snapshot."""
@dataclass(frozen=True)
class PersistedConfigSnapshot:
"""Validated configuration as represented by the persisted source."""
config: Config
path: Path
revision: str
@dataclass(frozen=True)
class EffectiveConfigSnapshot:
"""Runtime configuration with environment references resolved."""
config: Config
path: Path
revision: str
@dataclass(frozen=True)
class ConfigCommit:
"""One atomic configuration update and its semantic field changes."""
before: PersistedConfigSnapshot
after: PersistedConfigSnapshot
changed_paths: frozenset[str]
def _ensure_schema_refs() -> None:
global _schema_refs_ready
if _schema_refs_ready:
return
with _schema_refs_lock:
if not _schema_refs_ready:
_resolve_tool_config_refs()
_schema_refs_ready = True
def _lock_for_path(path: Path) -> threading.RLock:
key = path.expanduser().resolve(strict=False)
with _path_locks_guard:
return _path_locks.setdefault(key, threading.RLock())
def _revision_for_bytes(raw: bytes | None) -> str:
if raw is None:
return "missing"
return f"sha256:{hashlib.sha256(raw).hexdigest()}"
def _config_data(config: Config) -> dict[str, Any]:
data = config.model_dump(mode="json", by_alias=True)
if config.providers.openai_codex.proxy is not None:
data.setdefault("providers", {})["openaiCodex"] = {
"proxy": config.providers.openai_codex.proxy,
}
return data
def _validate_config_data(data: dict[str, Any], path: Path) -> Config:
_ensure_schema_refs()
try:
return Config.model_validate(data)
except ValueError as exc:
raise ValueError(f"Failed to load config from {path}: {exc}") from exc
def _read_snapshot(path: Path) -> PersistedConfigSnapshot:
_ensure_schema_refs()
if not path.exists():
return PersistedConfigSnapshot(Config(), path, "missing")
try:
raw = path.read_bytes()
data = json.loads(raw.decode("utf-8"))
if not isinstance(data, dict):
raise ValueError("config root must be a JSON object")
data = _migrate_config(data)
config = Config.model_validate(data)
except (UnicodeDecodeError, ValueError) as exc:
raise ValueError(f"Failed to load config from {path}: {exc}") from exc
return PersistedConfigSnapshot(config, path, _revision_for_bytes(raw))
def _write_config_atomic(path: Path, data: dict[str, Any]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
content = json.dumps(data, indent=2, ensure_ascii=False)
tmp = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp")
existing_mode = stat.S_IMODE(path.stat().st_mode) if path.exists() else None
try:
with open(tmp, "w", encoding="utf-8") as handle:
handle.write(content)
if existing_mode is not None:
with suppress(OSError):
os.chmod(tmp, existing_mode)
handle.flush()
os.fsync(handle.fileno())
os.replace(tmp, path)
with suppress(OSError, NotImplementedError):
directory_fd = os.open(path.parent, os.O_RDONLY)
try:
os.fsync(directory_fd)
finally:
os.close(directory_fd)
finally:
tmp.unlink(missing_ok=True)
def _changed_paths(before: Any, after: Any, prefix: str = "") -> set[str]:
if isinstance(before, dict) and isinstance(after, dict):
changed: set[str] = set()
for key in before.keys() | after.keys():
path = f"{prefix}.{key}" if prefix else str(key)
if key not in before or key not in after:
changed.add(path)
else:
changed.update(_changed_paths(before[key], after[key], path))
return changed
if before != after:
return {prefix} if prefix else {"<root>"}
return set()
class FileConfigRepository:
"""Read and atomically update one configuration file.
The repository does not cache. Every read returns a new validated snapshot,
while updates for the same path are serialized within this process.
"""
def __init__(self, path: str | Path):
self.path = Path(path).expanduser().resolve(strict=False)
self._lock = _lock_for_path(self.path)
def load_raw(self) -> PersistedConfigSnapshot:
"""Load the persisted form without resolving secret references."""
with self._lock:
return _read_snapshot(self.path)
def load_effective(self) -> EffectiveConfigSnapshot:
"""Load an isolated runtime snapshot with ``${VAR}`` references resolved."""
raw = self.load_raw()
effective = resolve_config_env_vars(raw.config.model_copy(deep=True))
return EffectiveConfigSnapshot(effective, raw.path, raw.revision)
def save(
self,
config: Config,
*,
expected_revision: str | None = None,
) -> PersistedConfigSnapshot:
"""Atomically save a complete config, optionally rejecting stale writes."""
with self._lock:
current = _read_snapshot(self.path)
if expected_revision is not None and current.revision != expected_revision:
raise ConfigConflictError(
f"Config changed since revision {expected_revision}; "
f"current revision is {current.revision}"
)
data = _config_data(config)
_validate_config_data(data, self.path)
_write_config_atomic(self.path, data)
return _read_snapshot(self.path)
def update(
self,
mutator: ConfigMutator,
*,
expected_revision: str | None = None,
) -> ConfigCommit:
"""Atomically apply a mutation to the latest persisted config."""
with self._lock:
before = _read_snapshot(self.path)
if expected_revision is not None and before.revision != expected_revision:
raise ConfigConflictError(
f"Config changed since revision {expected_revision}; "
f"current revision is {before.revision}"
)
before_data = _config_data(before.config)
draft = before.config.model_copy(deep=True)
mutator(draft)
after_data = _config_data(draft)
changed = frozenset(_changed_paths(before_data, after_data))
if not changed:
return ConfigCommit(before, before, changed)
_validate_config_data(after_data, self.path)
_write_config_atomic(self.path, after_data)
after = _read_snapshot(self.path)
return ConfigCommit(before, after, changed)
def resolve_config_env_vars(config: Config) -> Config:
"""Return *config* with ``${VAR}`` environment references resolved."""
return _resolve_in_place(config)
def _resolve_in_place(obj: Any) -> Any:
if isinstance(obj, str):
new = _ENV_REF_PATTERN.sub(_env_replace, obj)
return new if new != obj else obj
if isinstance(obj, BaseModel):
updates: dict[str, Any] = {}
for name in type(obj).model_fields:
old = getattr(obj, name)
new = _resolve_in_place(old)
if new is not old:
updates[name] = new
extras = obj.__pydantic_extra__
new_extras: dict[str, Any] | None = None
if extras:
resolved = {key: _resolve_in_place(value) for key, value in extras.items()}
if any(resolved[key] is not extras[key] for key in extras):
new_extras = resolved
if not updates and new_extras is None:
return obj
copy = obj.model_copy(update=updates) if updates else obj.model_copy()
if new_extras is not None:
copy.__pydantic_extra__ = new_extras
return copy
if isinstance(obj, dict):
resolved = {key: _resolve_in_place(value) for key, value in obj.items()}
return resolved if any(resolved[key] is not obj[key] for key in obj) else obj
if isinstance(obj, list):
resolved = [_resolve_in_place(value) for value in obj]
return resolved if any(new is not old for new, old in zip(resolved, obj)) else obj
return obj
def _resolve_env_vars(obj: object) -> object:
"""Recursively resolve ``${VAR}`` patterns in plain strings/dicts/lists."""
if isinstance(obj, str):
return _ENV_REF_PATTERN.sub(_env_replace, obj)
if isinstance(obj, dict):
return {key: _resolve_env_vars(value) for key, value in obj.items()}
if isinstance(obj, list):
return [_resolve_env_vars(value) for value in obj]
return obj
def _env_replace(match: re.Match[str]) -> str:
name = match.group(1)
value = os.environ.get(name)
if value is None:
raise ValueError(f"Environment variable '{name}' referenced in config is not set")
return value
def _migrate_config(data: dict[str, Any]) -> dict[str, Any]:
"""Migrate old config formats to the current schema."""
agents = data.get("agents", {})
defaults = agents.get("defaults", {}) if isinstance(agents, dict) else {}
if isinstance(defaults, dict):
had_legacy_max_messages = "maxMessages" in defaults or "max_messages" in defaults
defaults.pop("maxMessages", None)
defaults.pop("max_messages", None)
if had_legacy_max_messages:
logger.warning(
"agents.defaults.maxMessages/max_messages is legacy and ignored; "
"replay max messages is now an internal safety cap. Remove it from "
"config. This compatibility warning will be removed in the next version."
)
tools = data.get("tools", {})
if not isinstance(tools, dict):
return data
exec_cfg = tools.get("exec", {})
if isinstance(exec_cfg, dict) and "restrictToWorkspace" in exec_cfg and "restrictToWorkspace" not in tools:
tools["restrictToWorkspace"] = exec_cfg.pop("restrictToWorkspace")
if "myEnabled" in tools or "mySet" in tools:
my_cfg = tools.setdefault("my", {})
if not isinstance(my_cfg, dict):
return data
if "myEnabled" in tools and "enable" not in my_cfg:
my_cfg["enable"] = tools.pop("myEnabled")
else:
tools.pop("myEnabled", None)
if "mySet" in tools and "allowSet" not in my_cfg:
my_cfg["allowSet"] = tools.pop("mySet")
else:
tools.pop("mySet", None)
return data

9
nanobot/nanobot.py
View File

@ -95,7 +95,10 @@ class Nanobot:
model: Override the instance default model.
model_preset: Override the instance default model preset.
"""
from nanobot.config.loader import load_config, resolve_config_env_vars
from nanobot.config.loader import (
apply_config_runtime_policies,
load_effective_config,
)
ensure_single_model_selector(model=model, model_preset=model_preset)
resolved: Path | None = None
@ -104,7 +107,7 @@ class Nanobot:
if not resolved.exists():
raise FileNotFoundError(f"Config not found: {resolved}")
config: Config = resolve_config_env_vars(load_config(resolved))
config: Config = load_effective_config(resolved)
if workspace is not None:
config.agents.defaults.workspace = str(
Path(workspace).expanduser().resolve()
@ -116,6 +119,8 @@ class Nanobot:
elif model_preset is not None:
config.agents.defaults.model_preset = model_preset
apply_config_runtime_policies(config)
loop = AgentLoop.from_config(
config,
image_generation_provider_configs=image_gen_provider_configs(config),

87
nanobot/optional_features.py
View File

@ -26,7 +26,7 @@ from nanobot.channels._setup import (
stringify_channel_value,
)
from nanobot.channels.registry import DEFAULT_ENABLED_CHANNELS
from nanobot.config.loader import merge_missing_defaults
from nanobot.config.loader import get_config_repository, merge_missing_defaults
from nanobot.config.schema import Config
@ -266,29 +266,16 @@ def install_extra(
return InstallResult(False, label, pip_cmd, failed_cmd=failed_cmd, output=output)
def read_config_data(path: Path) -> dict[str, Any]:
if not path.exists():
return {}
with open(path, encoding="utf-8") as f:
return json.load(f)
def write_config_data(path: Path, data: dict[str, Any]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2, ensure_ascii=False)
def enable_channel_config(config_path: Path, channel_name: str, defaults: dict[str, Any]) -> None:
data = read_config_data(config_path)
channels = data.setdefault("channels", {})
existing = channels.get(channel_name, {})
if not isinstance(existing, dict):
existing = {}
merged = merge_missing_defaults(existing, defaults)
merged["enabled"] = True
channels[channel_name] = merged
write_config_data(config_path, data)
def mutate(config: Config) -> None:
existing = getattr(config.channels, channel_name, {})
if not isinstance(existing, dict):
existing = {}
merged = merge_missing_defaults(existing, defaults)
merged["enabled"] = True
setattr(config.channels, channel_name, merged)
get_config_repository(config_path).update(mutate)
def enable_feishu_instance_config(
@ -297,24 +284,29 @@ def enable_feishu_instance_config(
*,
instance_id: str = DEFAULT_INSTANCE_ID,
) -> None:
data = read_config_data(config_path)
channels = data.setdefault("channels", {})
existing = channels.get("feishu", {})
if not isinstance(existing, dict):
existing = {}
channels["feishu"] = set_feishu_instance_enabled(existing, defaults, instance_id, True)
write_config_data(config_path, data)
def mutate(config: Config) -> None:
existing = getattr(config.channels, "feishu", {})
if not isinstance(existing, dict):
existing = {}
config.channels.feishu = set_feishu_instance_enabled(
existing,
defaults,
instance_id,
True,
)
get_config_repository(config_path).update(mutate)
def disable_channel_config(config_path: Path, channel_name: str) -> None:
data = read_config_data(config_path)
channels = data.setdefault("channels", {})
existing = channels.get(channel_name, {})
if not isinstance(existing, dict):
existing = {}
existing["enabled"] = False
channels[channel_name] = existing
write_config_data(config_path, data)
def mutate(config: Config) -> None:
existing = getattr(config.channels, channel_name, {})
if not isinstance(existing, dict):
existing = {}
existing["enabled"] = False
setattr(config.channels, channel_name, existing)
get_config_repository(config_path).update(mutate)
def disable_feishu_instance_config(
@ -323,13 +315,18 @@ def disable_feishu_instance_config(
*,
instance_id: str = DEFAULT_INSTANCE_ID,
) -> None:
data = read_config_data(config_path)
channels = data.setdefault("channels", {})
existing = channels.get("feishu", {})
if not isinstance(existing, dict):
existing = {}
channels["feishu"] = set_feishu_instance_enabled(existing, defaults, instance_id, False)
write_config_data(config_path, data)
def mutate(config: Config) -> None:
existing = getattr(config.channels, "feishu", {})
if not isinstance(existing, dict):
existing = {}
config.channels.feishu = set_feishu_instance_enabled(
existing,
defaults,
instance_id,
False,
)
get_config_repository(config_path).update(mutate)
def channel_enabled(config: Config, name: str) -> bool:

59
nanobot/webui/mcp_presets_api.py
View File

@ -18,9 +18,9 @@ from typing import Any, Literal, Mapping
from nanobot.agent.tools.registry import ToolRegistry
from nanobot.apps.protocol import app_manifest, compact_dict
from nanobot.config.loader import load_config, resolve_config_env_vars, save_config
from nanobot.config.loader import load_config, resolve_config_env_vars, update_config
from nanobot.config.paths import get_runtime_subdir
from nanobot.config.schema import MCPServerConfig
from nanobot.config.schema import Config, MCPServerConfig
from nanobot.utils.helpers import ensure_dir
QueryParams = dict[str, list[str]]
@ -1180,19 +1180,16 @@ def _import_mcp_servers(raw_json: str | None) -> dict[str, MCPServerConfig]:
def custom_mcp_action(action: str, query: QueryParams) -> dict[str, Any]:
config = load_config()
if action == "custom":
name, cfg = _custom_server_from_query(query)
config.tools.mcp_servers[name] = cfg
save_config(config)
update_config(lambda config: config.tools.mcp_servers.__setitem__(name, cfg))
payload = mcp_presets_payload(last_action=_server_action_message(action, name))
payload["requires_restart"] = True
return payload
if action in {"import", "import-cursor"}:
servers = _import_mcp_servers(_query_first(query, "config"))
config.tools.mcp_servers.update(servers)
save_config(config)
update_config(lambda config: config.tools.mcp_servers.update(servers))
payload = mcp_presets_payload(last_action={
"ok": True,
"message": f"Imported {len(servers)} MCP server(s).",
@ -1202,12 +1199,15 @@ def custom_mcp_action(action: str, query: QueryParams) -> dict[str, Any]:
if action == "tools":
name = _validated_server_name((_query_first(query, "name") or "").strip())
cfg = config.tools.mcp_servers.get(name)
if cfg is None:
raise McpPresetError("unknown MCP server", status=404)
cfg.enabled_tools = _parse_enabled_tools(_query_first(query, "enabled_tools"))
config.tools.mcp_servers[name] = cfg
save_config(config)
def mutate(config: Config) -> None:
cfg = config.tools.mcp_servers.get(name)
if cfg is None:
raise McpPresetError("unknown MCP server", status=404)
cfg.enabled_tools = _parse_enabled_tools(_query_first(query, "enabled_tools"))
config.tools.mcp_servers[name] = cfg
update_config(mutate)
payload = mcp_presets_payload(last_action=_server_action_message(action, name))
payload["requires_restart"] = True
return payload
@ -1221,31 +1221,42 @@ def mcp_presets_action(action: str, query: QueryParams) -> dict[str, Any]:
raise McpPresetError("missing MCP preset name")
preset = _preset_by_name_optional(name)
config = load_config()
existing = config.tools.mcp_servers.get(name)
if action == "enable":
if preset is None:
raise McpPresetError("unknown MCP preset", status=404)
config.tools.mcp_servers[preset.name] = _materialize_server(preset, query, existing)
save_config(config)
def mutate(config: Config) -> None:
existing = config.tools.mcp_servers.get(name)
config.tools.mcp_servers[preset.name] = _materialize_server(
preset,
query,
existing,
)
update_config(mutate)
payload = mcp_presets_payload(last_action=_action_message(action, preset))
payload["requires_restart"] = True
return payload
if action == "remove":
if preset is None and name not in config.tools.mcp_servers:
raise McpPresetError("unknown MCP server", status=404)
removed_runtime_files = False
cleanup_error = ""
if name in config.tools.mcp_servers:
existing_cfg = config.tools.mcp_servers[name]
removed: dict[str, MCPServerConfig] = {}
def mutate(config: Config) -> None:
existing_cfg = config.tools.mcp_servers.get(name)
if preset is None and existing_cfg is None:
raise McpPresetError("unknown MCP server", status=404)
if existing_cfg is not None:
removed[name] = existing_cfg
del config.tools.mcp_servers[name]
update_config(mutate)
if existing_cfg := removed.get(name):
try:
removed_runtime_files = _remove_managed_stdio_cwd(name, existing_cfg)
except OSError as exc:
cleanup_error = str(exc)
del config.tools.mcp_servers[name]
save_config(config)
last_action = (
_action_message(action, preset)
if preset is not None

552
nanobot/webui/settings_api.py
View File

@ -22,8 +22,13 @@ from nanobot.audio.transcription_registry import (
resolve_transcription_provider,
transcription_provider_names,
)
from nanobot.config.loader import get_config_path, load_config, resolve_config_env_vars, save_config
from nanobot.config.schema import ModelPresetConfig, ProviderConfig
from nanobot.config.loader import (
get_config_path,
load_config,
resolve_config_env_vars,
update_config,
)
from nanobot.config.schema import Config, ModelPresetConfig, ProviderConfig
from nanobot.providers.image_generation import (
get_image_gen_provider,
image_gen_provider_names,
@ -961,100 +966,86 @@ def settings_usage_payload() -> dict[str, Any]:
def update_agent_settings(query: QueryParams) -> dict[str, Any]:
config = load_config()
defaults = config.agents.defaults
changed = False
restart_required = False
if "model_preset" in query or "modelPreset" in query:
preset = (_query_first_alias(query, "model_preset", "modelPreset") or "").strip()
preset_value = None if not preset or preset == "default" else preset
if preset_value is not None and preset_value not in config.model_presets:
raise WebUISettingsError("unknown model preset")
if defaults.model_preset != preset_value:
def mutate(config: Config) -> None:
nonlocal restart_required
defaults = config.agents.defaults
if "model_preset" in query or "modelPreset" in query:
preset = (_query_first_alias(query, "model_preset", "modelPreset") or "").strip()
preset_value = None if not preset or preset == "default" else preset
if preset_value is not None and preset_value not in config.model_presets:
raise WebUISettingsError("unknown model preset")
defaults.model_preset = preset_value
changed = True
model = _query_first(query, "model")
if model is not None:
model = model.strip()
if not model:
raise WebUISettingsError("model is required")
if defaults.model != model:
model = _query_first(query, "model")
if model is not None:
model = model.strip()
if not model:
raise WebUISettingsError("model is required")
defaults.model = model
changed = True
provider = _query_first(query, "provider")
if provider is not None:
provider = provider.strip()
if not provider:
raise WebUISettingsError("provider is required")
_validate_configured_provider(config, provider)
if defaults.provider != provider:
provider = _query_first(query, "provider")
if provider is not None:
provider = provider.strip()
if not provider:
raise WebUISettingsError("provider is required")
_validate_configured_provider(config, provider)
defaults.provider = provider
changed = True
context_window_tokens = _parse_context_window_tokens(
_query_first_alias(query, "context_window_tokens", "contextWindowTokens")
)
if (
context_window_tokens is not None
and defaults.context_window_tokens != context_window_tokens
):
defaults.context_window_tokens = context_window_tokens
changed = True
context_window_tokens = _parse_context_window_tokens(
_query_first_alias(query, "context_window_tokens", "contextWindowTokens")
)
if context_window_tokens is not None:
defaults.context_window_tokens = context_window_tokens
timezone = _query_first(query, "timezone")
if timezone is not None:
timezone = timezone.strip()
if not timezone:
raise WebUISettingsError("timezone is required")
try:
ZoneInfo(timezone)
except Exception:
raise WebUISettingsError("invalid timezone") from None
if defaults.timezone != timezone:
defaults.timezone = timezone
changed = True
restart_required = True
timezone = _query_first(query, "timezone")
if timezone is not None:
timezone = timezone.strip()
if not timezone:
raise WebUISettingsError("timezone is required")
try:
ZoneInfo(timezone)
except Exception:
raise WebUISettingsError("invalid timezone") from None
if defaults.timezone != timezone:
defaults.timezone = timezone
restart_required = True
bot_name = _query_first_alias(query, "bot_name", "botName")
if bot_name is not None:
bot_name = bot_name.strip()
if not bot_name:
raise WebUISettingsError("bot_name is required")
if defaults.bot_name != bot_name:
defaults.bot_name = bot_name
changed = True
restart_required = True
bot_name = _query_first_alias(query, "bot_name", "botName")
if bot_name is not None:
bot_name = bot_name.strip()
if not bot_name:
raise WebUISettingsError("bot_name is required")
if defaults.bot_name != bot_name:
defaults.bot_name = bot_name
restart_required = True
bot_icon = _query_first_alias(query, "bot_icon", "botIcon")
if bot_icon is not None:
bot_icon = bot_icon.strip()
if defaults.bot_icon != bot_icon:
defaults.bot_icon = bot_icon
changed = True
restart_required = True
bot_icon = _query_first_alias(query, "bot_icon", "botIcon")
if bot_icon is not None:
bot_icon = bot_icon.strip()
if defaults.bot_icon != bot_icon:
defaults.bot_icon = bot_icon
restart_required = True
tool_hint_max_length = _query_first_alias(
query,
"tool_hint_max_length",
"toolHintMaxLength",
)
if tool_hint_max_length is not None:
try:
parsed = int(tool_hint_max_length)
except ValueError:
raise WebUISettingsError("tool_hint_max_length must be an integer") from None
if parsed < 20 or parsed > 500:
raise WebUISettingsError("tool_hint_max_length must be between 20 and 500")
if defaults.tool_hint_max_length != parsed:
defaults.tool_hint_max_length = parsed
changed = True
restart_required = True
tool_hint_max_length = _query_first_alias(
query,
"tool_hint_max_length",
"toolHintMaxLength",
)
if tool_hint_max_length is not None:
try:
parsed = int(tool_hint_max_length)
except ValueError:
raise WebUISettingsError("tool_hint_max_length must be an integer") from None
if parsed < 20 or parsed > 500:
raise WebUISettingsError("tool_hint_max_length must be between 20 and 500")
if defaults.tool_hint_max_length != parsed:
defaults.tool_hint_max_length = parsed
restart_required = True
if changed:
save_config(config)
update_config(mutate)
return settings_payload(requires_restart=restart_required)
@ -1072,23 +1063,23 @@ def create_model_configuration(query: QueryParams) -> dict[str, Any]:
raise WebUISettingsError("provider is required")
name = _model_configuration_slug(raw_name or label)
config = load_config()
if name in config.model_presets:
raise WebUISettingsError("configuration already exists", status=409)
_validate_configured_provider(config, provider)
def mutate(config: Config) -> None:
if name in config.model_presets:
raise WebUISettingsError("configuration already exists", status=409)
_validate_configured_provider(config, provider)
base = config.resolve_default_preset()
config.model_presets[name] = ModelPresetConfig(
label=label,
model=model,
provider=provider,
max_tokens=base.max_tokens,
context_window_tokens=base.context_window_tokens,
temperature=base.temperature,
reasoning_effort=base.reasoning_effort,
)
config.agents.defaults.model_preset = name
base = config.resolve_default_preset()
config.model_presets[name] = ModelPresetConfig(
label=label,
model=model,
provider=provider,
max_tokens=base.max_tokens,
context_window_tokens=base.context_window_tokens,
temperature=base.temperature,
reasoning_effort=base.reasoning_effort,
)
config.agents.defaults.model_preset = name
save_config(config)
update_config(mutate)
return settings_payload()
@ -1097,56 +1088,41 @@ def update_model_configuration(query: QueryParams) -> dict[str, Any]:
if not name or name == "default":
raise WebUISettingsError("model configuration is required")
config = load_config()
preset = config.model_presets.get(name)
if preset is None:
raise WebUISettingsError("unknown model configuration")
def mutate(config: Config) -> None:
preset = config.model_presets.get(name)
if preset is None:
raise WebUISettingsError("unknown model configuration")
changed = False
label = _query_first_alias(query, "label", "displayName")
if label is not None:
label = label.strip()
if not label:
raise WebUISettingsError("label is required")
if preset.label != label:
label = _query_first_alias(query, "label", "displayName")
if label is not None:
label = label.strip()
if not label:
raise WebUISettingsError("label is required")
preset.label = label
changed = True
model = _query_first(query, "model")
if model is not None:
model = model.strip()
if not model:
raise WebUISettingsError("model is required")
if preset.model != model:
model = _query_first(query, "model")
if model is not None:
model = model.strip()
if not model:
raise WebUISettingsError("model is required")
preset.model = model
changed = True
provider = _query_first(query, "provider")
if provider is not None:
provider = provider.strip()
if not provider:
raise WebUISettingsError("provider is required")
_validate_configured_provider(config, provider)
if preset.provider != provider:
provider = _query_first(query, "provider")
if provider is not None:
provider = provider.strip()
if not provider:
raise WebUISettingsError("provider is required")
_validate_configured_provider(config, provider)
preset.provider = provider
changed = True
context_window_tokens = _parse_context_window_tokens(
_query_first_alias(query, "context_window_tokens", "contextWindowTokens")
)
if (
context_window_tokens is not None
and preset.context_window_tokens != context_window_tokens
):
preset.context_window_tokens = context_window_tokens
changed = True
if config.agents.defaults.model_preset != name:
context_window_tokens = _parse_context_window_tokens(
_query_first_alias(query, "context_window_tokens", "contextWindowTokens")
)
if context_window_tokens is not None:
preset.context_window_tokens = context_window_tokens
config.agents.defaults.model_preset = name
changed = True
if changed:
save_config(config)
update_config(mutate)
return settings_payload()
@ -1155,43 +1131,37 @@ def update_provider_settings(query: QueryParams) -> dict[str, Any]:
if not provider_name:
raise WebUISettingsError("provider is required")
config = load_config()
resolved_provider = _resolve_settings_provider(config, provider_name)
if resolved_provider is None:
raise WebUISettingsError("unknown provider")
spec, provider_key, provider_config = resolved_provider
if spec.is_oauth:
raise WebUISettingsError("unknown provider")
provider_key = ""
changed = False
if "api_key" in query or "apiKey" in query:
api_key = _query_first_alias(query, "api_key", "apiKey")
api_key = (api_key or "").strip() or None
if provider_config.api_key != api_key:
provider_config.api_key = api_key
changed = True
def mutate(config: Config) -> None:
nonlocal provider_key
resolved_provider = _resolve_settings_provider(config, provider_name)
if resolved_provider is None:
raise WebUISettingsError("unknown provider")
spec, provider_key, provider_config = resolved_provider
if spec.is_oauth:
raise WebUISettingsError("unknown provider")
if "api_base" in query or "apiBase" in query:
api_base = _query_first_alias(query, "api_base", "apiBase")
api_base = (api_base or "").strip() or None
if provider_config.api_base != api_base:
provider_config.api_base = api_base
changed = True
if "api_key" in query or "apiKey" in query:
api_key = _query_first_alias(query, "api_key", "apiKey")
provider_config.api_key = (api_key or "").strip() or None
if "api_type" in query:
if spec.name == "openai":
if "api_base" in query or "apiBase" in query:
api_base = _query_first_alias(query, "api_base", "apiBase")
provider_config.api_base = (api_base or "").strip() or None
if "api_type" in query and spec.name == "openai":
api_type = (_query_first(query, "api_type") or "").strip()
try:
parsed_api_type = type(provider_config)(api_type=api_type).api_type
provider_config.api_type = type(provider_config)(api_type=api_type).api_type
except Exception:
raise WebUISettingsError("api_type must be auto, chat_completions, or responses") from None
if provider_config.api_type != parsed_api_type:
provider_config.api_type = parsed_api_type
changed = True
raise WebUISettingsError(
"api_type must be auto, chat_completions, or responses"
) from None
if changed:
save_config(config)
image_config = config.tools.image_generation
commit = update_config(mutate)
changed = bool(commit.changed_paths)
image_config = commit.after.config.tools.image_generation
restart_required = (
changed
and image_config.enabled
@ -1299,16 +1269,18 @@ def update_network_safety_settings(query: QueryParams) -> dict[str, Any]:
if raw_allow is None and raw_default_access_mode is None:
raise WebUISettingsError("webui_allow_local_service_access or webui_default_access_mode is required")
config = load_config()
changed = False
if raw_allow is not None:
webui_allow_local_service_access = _parse_bool(raw_allow, "webui_allow_local_service_access")
if config.tools.webui_allow_local_service_access != webui_allow_local_service_access:
config.tools.webui_allow_local_service_access = webui_allow_local_service_access
changed = True
if changed:
save_config(config)
commit = update_config(
lambda config: setattr(
config.tools,
"webui_allow_local_service_access",
webui_allow_local_service_access,
)
)
changed = bool(commit.changed_paths)
else:
changed = False
if raw_default_access_mode is not None:
default_access_mode = raw_default_access_mode.strip().lower()
if default_access_mode == "restricted":
@ -1328,134 +1300,120 @@ def update_web_search_settings(query: QueryParams) -> dict[str, Any]:
if provider_option is None:
raise WebUISettingsError("unknown web search provider")
config = load_config()
search_config = config.tools.web.search
web_config = config.tools.web
previous_provider = search_config.provider
changed = False
restart_required = False
def set_search_value(attr: str, value: object) -> None:
nonlocal changed
if getattr(search_config, attr) != value:
setattr(search_config, attr, value)
changed = True
def set_fetch_value(attr: str, value: object) -> None:
nonlocal changed
if getattr(web_config.fetch, attr) != value:
setattr(web_config.fetch, attr, value)
changed = True
if search_config.provider != provider_name:
def mutate(config: Config) -> None:
nonlocal restart_required
search_config = config.tools.web.search
web_config = config.tools.web
previous_provider = search_config.provider
search_config.provider = provider_name
changed = True
credential = provider_option["credential"]
if credential == "none":
set_search_value("api_key", "")
set_search_value("base_url", "")
elif credential == "base_url":
base_url = _query_first_alias(query, "base_url", "baseUrl")
base_url = base_url.strip() if base_url is not None else None
if not base_url and previous_provider == provider_name and search_config.base_url:
base_url = search_config.base_url
if not base_url:
raise WebUISettingsError("base_url is required")
set_search_value("base_url", base_url)
set_search_value("api_key", "")
elif credential in {"api_key", "optional_api_key"}:
raw_api_key = _query_first_alias(query, "api_key", "apiKey")
api_key = raw_api_key.strip() if raw_api_key is not None else None
if api_key is None and previous_provider == provider_name and search_config.api_key:
api_key = search_config.api_key
if credential == "api_key" and not api_key:
raise WebUISettingsError("api_key is required")
set_search_value("api_key", api_key or "")
set_search_value("base_url", "")
else:
raise WebUISettingsError("unknown web search credential type")
credential = provider_option["credential"]
if credential == "none":
search_config.api_key = ""
search_config.base_url = ""
elif credential == "base_url":
base_url = _query_first_alias(query, "base_url", "baseUrl")
base_url = base_url.strip() if base_url is not None else None
if not base_url and previous_provider == provider_name and search_config.base_url:
base_url = search_config.base_url
if not base_url:
raise WebUISettingsError("base_url is required")
search_config.base_url = base_url
search_config.api_key = ""
elif credential in {"api_key", "optional_api_key"}:
raw_api_key = _query_first_alias(query, "api_key", "apiKey")
api_key = raw_api_key.strip() if raw_api_key is not None else None
if api_key is None and previous_provider == provider_name and search_config.api_key:
api_key = search_config.api_key
if credential == "api_key" and not api_key:
raise WebUISettingsError("api_key is required")
search_config.api_key = api_key or ""
search_config.base_url = ""
else:
raise WebUISettingsError("unknown web search credential type")
max_results = _query_first_alias(query, "max_results", "maxResults")
if max_results is not None:
try:
parsed = int(max_results)
except ValueError:
raise WebUISettingsError("max_results must be an integer") from None
if parsed < 1 or parsed > 10:
raise WebUISettingsError("max_results must be between 1 and 10")
set_search_value("max_results", parsed)
max_results = _query_first_alias(query, "max_results", "maxResults")
if max_results is not None:
try:
parsed = int(max_results)
except ValueError:
raise WebUISettingsError("max_results must be an integer") from None
if parsed < 1 or parsed > 10:
raise WebUISettingsError("max_results must be between 1 and 10")
search_config.max_results = parsed
timeout = _query_first(query, "timeout")
if timeout is not None:
try:
parsed_timeout = int(timeout)
except ValueError:
raise WebUISettingsError("timeout must be an integer") from None
if parsed_timeout < 1 or parsed_timeout > 120:
raise WebUISettingsError("timeout must be between 1 and 120")
set_search_value("timeout", parsed_timeout)
timeout = _query_first(query, "timeout")
if timeout is not None:
try:
parsed_timeout = int(timeout)
except ValueError:
raise WebUISettingsError("timeout must be an integer") from None
if parsed_timeout < 1 or parsed_timeout > 120:
raise WebUISettingsError("timeout must be between 1 and 120")
search_config.timeout = parsed_timeout
use_jina_reader = _query_first_alias(query, "use_jina_reader", "useJinaReader")
if use_jina_reader is not None:
normalized = use_jina_reader.strip().lower()
if normalized not in {"1", "0", "true", "false", "yes", "no"}:
raise WebUISettingsError("use_jina_reader must be boolean")
previous_jina_reader = web_config.fetch.use_jina_reader
set_fetch_value("use_jina_reader", normalized in {"1", "true", "yes"})
if web_config.fetch.use_jina_reader != previous_jina_reader:
restart_required = True
use_jina_reader = _query_first_alias(query, "use_jina_reader", "useJinaReader")
if use_jina_reader is not None:
normalized = use_jina_reader.strip().lower()
if normalized not in {"1", "0", "true", "false", "yes", "no"}:
raise WebUISettingsError("use_jina_reader must be boolean")
enabled = normalized in {"1", "true", "yes"}
if web_config.fetch.use_jina_reader != enabled:
web_config.fetch.use_jina_reader = enabled
restart_required = True
if changed:
save_config(config)
update_config(mutate)
return settings_payload(requires_restart=restart_required)
def update_api_settings(query: QueryParams) -> dict[str, Any]:
"""Update the managed OpenAI-compatible API configuration."""
config = load_config()
api = config.api
def mutate(config: Config) -> None:
api = config.api
host = _query_first(query, "host")
if host is not None:
host = host.strip()
if not host:
raise WebUISettingsError("host is required")
api.host = host
host = _query_first(query, "host")
if host is not None:
host = host.strip()
if not host:
raise WebUISettingsError("host is required")
api.host = host
port = _query_first(query, "port")
if port is not None:
try:
parsed_port = int(port)
except ValueError:
raise WebUISettingsError("port must be an integer") from None
if parsed_port < 1 or parsed_port > 65535:
raise WebUISettingsError("port must be between 1 and 65535")
api.port = parsed_port
port = _query_first(query, "port")
if port is not None:
try:
parsed_port = int(port)
except ValueError:
raise WebUISettingsError("port must be an integer") from None
if parsed_port < 1 or parsed_port > 65535:
raise WebUISettingsError("port must be between 1 and 65535")
api.port = parsed_port
timeout = _query_first(query, "timeout")
if timeout is not None:
try:
parsed_timeout = float(timeout)
except ValueError:
raise WebUISettingsError("timeout must be a number") from None
if parsed_timeout < 1 or parsed_timeout > 3600:
raise WebUISettingsError("timeout must be between 1 and 3600")
api.timeout = parsed_timeout
timeout = _query_first(query, "timeout")
if timeout is not None:
try:
parsed_timeout = float(timeout)
except ValueError:
raise WebUISettingsError("timeout must be a number") from None
if parsed_timeout < 1 or parsed_timeout > 3600:
raise WebUISettingsError("timeout must be between 1 and 3600")
api.timeout = parsed_timeout
api_key = _query_first_alias(query, "api_key", "apiKey")
if api_key is not None:
api.api_key = api_key.strip()
api_key = _query_first_alias(query, "api_key", "apiKey")
if api_key is not None:
api.api_key = api_key.strip()
if not is_loopback_host(api.host) and not api.api_key.strip():
raise WebUISettingsError("an API key is required when the API is available on the network")
if not is_loopback_host(api.host) and not api.api_key.strip():
raise WebUISettingsError(
"an API key is required when the API is available on the network"
)
save_config(config)
update_config(mutate)
return settings_payload()
def update_image_generation_settings(query: QueryParams) -> dict[str, Any]:
config = load_config()
def _apply_image_generation_settings(config: Config, query: QueryParams) -> bool:
image_config = config.tools.image_generation
changed = False
@ -1547,13 +1505,21 @@ def update_image_generation_settings(query: QueryParams) -> dict[str, Any]:
if not selected_provider or not selected_provider["configured"]:
raise WebUISettingsError("image generation provider is not configured")
if changed:
save_config(config)
return changed
def update_image_generation_settings(query: QueryParams) -> dict[str, Any]:
changed = False
def mutate(config: Config) -> None:
nonlocal changed
changed = _apply_image_generation_settings(config, query)
update_config(mutate)
return settings_payload(requires_restart=changed)
def update_transcription_settings(query: QueryParams) -> dict[str, Any]:
config = load_config()
def _apply_transcription_settings(config: Config, query: QueryParams) -> bool:
transcription = config.transcription
changed = False
@ -1617,6 +1583,12 @@ def update_transcription_settings(query: QueryParams) -> dict[str, Any]:
transcription.max_upload_mb = parsed_upload
changed = True
if changed:
save_config(config)
return changed
def update_transcription_settings(query: QueryParams) -> dict[str, Any]:
def mutate(config: Config) -> None:
_apply_transcription_settings(config, query)
update_config(mutate)
return settings_payload()

89
nanobot/webui/settings_routes.py
View File

@ -21,7 +21,7 @@ from nanobot.agent.tools.mcp import request_mcp_reload
from nanobot.api.runtime import ApiRuntime, ApiStartOptions, api_runtime_paths
from nanobot.bus.queue import MessageBus
from nanobot.channels._setup import channel_setup_spec
from nanobot.config.loader import get_config_path, load_config, save_config
from nanobot.config.loader import get_config_path, load_config, update_config
from nanobot.optional_features import (
OptionalFeatureError,
extra_installed,
@ -773,51 +773,56 @@ class WebUISettingsRouter:
if not raw_values:
return []
config = load_config()
section = getattr(config.channels, name, None)
if name == "feishu":
from nanobot.channels._feishu_instances import feishu_instance_specs
from nanobot.channels.feishu import FeishuChannel
specs = feishu_instance_specs(section, FeishuChannel.default_config())
selected = next((spec for spec in specs if spec.instance_id == instance_id), None)
channel_config = dict(selected.config) if selected is not None else {}
elif hasattr(section, "model_dump"):
channel_config = section.model_dump(mode="json", by_alias=True)
elif isinstance(section, dict):
channel_config = dict(section)
else:
channel_config = {}
saved: list[str] = []
prefix = f"channels.{name}."
for raw_key, raw_value in raw_values.items():
if not isinstance(raw_key, str) or not raw_key:
raise WebUISettingsError("channel settings payload contains an invalid key")
field = raw_key[len(prefix):] if raw_key.startswith(prefix) else raw_key
value_type = field_types.get(field)
if value_type is None:
raise WebUISettingsError(f"'{raw_key}' cannot be configured from WebUI")
value = self._coerce_channel_value(raw_key, raw_value, value_type)
if value is _SKIP_FIELD:
continue
self._assign_channel_config_value(channel_config, field, value)
saved.append(raw_key)
if name == "feishu":
from nanobot.channels._feishu_instances import upsert_feishu_instance
from nanobot.channels.feishu import FeishuChannel
def mutate(config: Any) -> None:
section = getattr(config.channels, name, None)
if name == "feishu":
from nanobot.channels._feishu_instances import feishu_instance_specs
from nanobot.channels.feishu import FeishuChannel
existing = getattr(config.channels, name, None)
channel_config = upsert_feishu_instance(
existing if isinstance(existing, dict) else {},
FeishuChannel.default_config(),
instance_id,
channel_config,
)
specs = feishu_instance_specs(section, FeishuChannel.default_config())
selected = next(
(spec for spec in specs if spec.instance_id == instance_id),
None,
)
channel_config = dict(selected.config) if selected is not None else {}
elif hasattr(section, "model_dump"):
channel_config = section.model_dump(mode="json", by_alias=True)
elif isinstance(section, dict):
channel_config = dict(section)
else:
channel_config = {}
setattr(config.channels, name, channel_config)
save_config(config)
prefix = f"channels.{name}."
for raw_key, raw_value in raw_values.items():
if not isinstance(raw_key, str) or not raw_key:
raise WebUISettingsError("channel settings payload contains an invalid key")
field = raw_key[len(prefix):] if raw_key.startswith(prefix) else raw_key
value_type = field_types.get(field)
if value_type is None:
raise WebUISettingsError(f"'{raw_key}' cannot be configured from WebUI")
value = self._coerce_channel_value(raw_key, raw_value, value_type)
if value is _SKIP_FIELD:
continue
self._assign_channel_config_value(channel_config, field, value)
saved.append(raw_key)
if name == "feishu":
from nanobot.channels._feishu_instances import upsert_feishu_instance
from nanobot.channels.feishu import FeishuChannel
existing = getattr(config.channels, name, None)
channel_config = upsert_feishu_instance(
existing if isinstance(existing, dict) else {},
FeishuChannel.default_config(),
instance_id,
channel_config,
)
setattr(config.channels, name, channel_config)
update_config(mutate)
return saved
@staticmethod

8
tests/config/test_config_load_errors.py
View File

@ -20,6 +20,14 @@ def test_load_config_invalid_json_fails_fast(tmp_path) -> None:
load_config(config_path)
def test_load_config_rejects_non_object_root(tmp_path) -> None:
config_path = tmp_path / "config.json"
config_path.write_text("[]", encoding="utf-8")
with pytest.raises(ValueError, match="config root must be a JSON object"):
load_config(config_path)
def test_load_config_invalid_schema_fails_fast(tmp_path) -> None:
config_path = tmp_path / "config.json"
config_path.write_text(

8
tests/config/test_config_migration.py
View File

@ -4,7 +4,7 @@ from unittest.mock import patch
import pytest
from nanobot.config.loader import load_config, save_config
from nanobot.config.loader import apply_config_runtime_policies, load_config, save_config
from nanobot.security.network import validate_url_target
@ -244,7 +244,7 @@ def test_new_my_tool_keys_take_precedence_over_legacy(tmp_path) -> None:
assert config.tools.my.allow_set is True
def test_load_config_resets_ssrf_whitelist_when_next_config_is_empty(tmp_path) -> None:
def test_runtime_policy_application_resets_ssrf_whitelist(tmp_path) -> None:
whitelisted = tmp_path / "whitelisted.json"
whitelisted.write_text(
json.dumps({"tools": {"ssrfWhitelist": ["100.64.0.0/10"]}}),
@ -253,12 +253,12 @@ def test_load_config_resets_ssrf_whitelist_when_next_config_is_empty(tmp_path) -
defaulted = tmp_path / "defaulted.json"
defaulted.write_text(json.dumps({}), encoding="utf-8")
load_config(whitelisted)
apply_config_runtime_policies(load_config(whitelisted))
with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve("ts.local", ["100.100.1.1"])):
ok, err = validate_url_target("http://ts.local/api")
assert ok, err
load_config(defaulted)
apply_config_runtime_policies(load_config(defaulted))
with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve("ts.local", ["100.100.1.1"])):
ok, _ = validate_url_target("http://ts.local/api")
assert not ok

191
tests/config/test_repository.py Normal file
View File

@ -0,0 +1,191 @@
import json
import os
import socket
import stat
import threading
from pathlib import Path
from unittest.mock import patch
import pytest
from nanobot.config.loader import load_config
from nanobot.config.repository import ConfigConflictError, FileConfigRepository
from nanobot.security.network import configure_ssrf_whitelist, validate_url_target
def _fake_resolve(host: str, results: list[str]):
def _resolver(hostname, port, family=0, type_=0):
if hostname == host:
return [
(socket.AF_INET, socket.SOCK_STREAM, 0, "", (ip, 0))
for ip in results
]
raise socket.gaierror(f"cannot resolve {hostname}")
return _resolver
def test_raw_and_effective_snapshots_keep_secret_templates_separate(
tmp_path: Path,
monkeypatch,
) -> None:
path = tmp_path / "config.json"
path.write_text(
json.dumps({"providers": {"groq": {"apiKey": "${GROQ_TOKEN}"}}}),
encoding="utf-8",
)
monkeypatch.setenv("GROQ_TOKEN", "resolved-secret")
repository = FileConfigRepository(path)
raw = repository.load_raw()
effective = repository.load_effective()
assert raw.config.providers.groq.api_key == "${GROQ_TOKEN}"
assert effective.config.providers.groq.api_key == "resolved-secret"
assert effective.config is not raw.config
assert effective.revision == raw.revision
def test_update_uses_latest_raw_config_and_reports_changed_paths(
tmp_path: Path,
monkeypatch,
) -> None:
path = tmp_path / "config.json"
path.write_text(
json.dumps({"providers": {"groq": {"apiKey": "${GROQ_TOKEN}"}}}),
encoding="utf-8",
)
monkeypatch.setenv("GROQ_TOKEN", "resolved-secret")
repository = FileConfigRepository(path)
commit = repository.update(
lambda config: setattr(config.agents.defaults, "timezone", "Asia/Shanghai")
)
saved = json.loads(path.read_text(encoding="utf-8"))
assert saved["providers"]["groq"]["apiKey"] == "${GROQ_TOKEN}"
assert saved["agents"]["defaults"]["timezone"] == "Asia/Shanghai"
assert commit.before.revision != commit.after.revision
assert commit.changed_paths == frozenset({"agents.defaults.timezone"})
def test_update_rejects_stale_expected_revision(tmp_path: Path) -> None:
path = tmp_path / "config.json"
repository = FileConfigRepository(path)
initial = repository.load_raw()
repository.update(lambda config: setattr(config.api, "port", 9001))
with pytest.raises(ConfigConflictError, match="Config changed"):
repository.update(
lambda config: setattr(config.api, "port", 9002),
expected_revision=initial.revision,
)
assert repository.load_raw().config.api.port == 9001
def test_repositories_for_same_path_serialize_updates(tmp_path: Path) -> None:
path = tmp_path / "config.json"
first = FileConfigRepository(path)
second = FileConfigRepository(path)
first_mutator_entered = threading.Event()
release_first = threading.Event()
second_mutator_entered = threading.Event()
def update_first() -> None:
def mutate(config):
first_mutator_entered.set()
assert release_first.wait(timeout=2)
config.api.port = 9001
first.update(mutate)
def update_second() -> None:
def mutate(config):
second_mutator_entered.set()
config.agents.defaults.timezone = "Asia/Shanghai"
second.update(mutate)
first_thread = threading.Thread(target=update_first)
second_thread = threading.Thread(target=update_second)
first_thread.start()
assert first_mutator_entered.wait(timeout=2)
second_thread.start()
assert not second_mutator_entered.wait(timeout=0.1)
release_first.set()
first_thread.join(timeout=2)
second_thread.join(timeout=2)
config = first.load_raw().config
assert config.api.port == 9001
assert config.agents.defaults.timezone == "Asia/Shanghai"
def test_repositories_for_different_paths_are_isolated(tmp_path: Path) -> None:
first_path = tmp_path / "first.json"
second_path = tmp_path / "second.json"
first = FileConfigRepository(first_path)
second = FileConfigRepository(second_path)
first.update(lambda config: setattr(config.api, "port", 9001))
second.update(lambda config: setattr(config.api, "port", 9002))
assert first.load_raw().config.api.port == 9001
assert second.load_raw().config.api.port == 9002
assert json.loads(first_path.read_text(encoding="utf-8"))["api"]["port"] == 9001
assert json.loads(second_path.read_text(encoding="utf-8"))["api"]["port"] == 9002
def test_noop_update_keeps_revision_and_does_not_rewrite(tmp_path: Path) -> None:
path = tmp_path / "config.json"
repository = FileConfigRepository(path)
repository.update(lambda config: setattr(config.api, "port", 9001))
before = repository.load_raw()
commit = repository.update(lambda config: setattr(config.api, "port", 9001))
assert commit.changed_paths == frozenset()
assert commit.before.revision == before.revision
assert commit.after.revision == before.revision
def test_atomic_save_keeps_previous_file_when_replace_fails(tmp_path: Path) -> None:
path = tmp_path / "config.json"
path.write_text('{"api": {"port": 9000}}', encoding="utf-8")
repository = FileConfigRepository(path)
with patch("nanobot.config.repository.os.replace", side_effect=OSError("replace failed")):
with pytest.raises(OSError, match="replace failed"):
repository.update(lambda config: setattr(config.api, "port", 9001))
assert json.loads(path.read_text(encoding="utf-8"))["api"]["port"] == 9000
assert list(tmp_path.glob(".config.json.*.tmp")) == []
@pytest.mark.skipif(os.name == "nt", reason="Windows does not expose POSIX file modes")
def test_atomic_save_preserves_existing_file_mode(tmp_path: Path) -> None:
path = tmp_path / "config.json"
path.write_text('{"api": {"port": 9000}}', encoding="utf-8")
path.chmod(0o600)
FileConfigRepository(path).update(lambda config: setattr(config.api, "port", 9001))
assert stat.S_IMODE(path.stat().st_mode) == 0o600
def test_loading_config_does_not_change_process_network_policy(tmp_path: Path) -> None:
path = tmp_path / "config.json"
path.write_text(json.dumps({"tools": {"ssrfWhitelist": []}}), encoding="utf-8")
configure_ssrf_whitelist(["100.64.0.0/10"])
try:
load_config(path)
with patch(
"nanobot.security.network.socket.getaddrinfo",
_fake_resolve("ts.local", ["100.100.1.1"]),
):
ok, error = validate_url_target("http://ts.local/api")
assert ok, error
finally:
configure_ssrf_whitelist([])