136 lines
4.3 KiB
Python
136 lines
4.3 KiB
Python
"""Configuration loading utilities."""
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
from pathlib import Path
|
|
|
|
import pydantic
|
|
from loguru import logger
|
|
|
|
from nanobot.config.schema import Config
|
|
|
|
# Global variable to store current config path (for multi-instance support)
|
|
_current_config_path: Path | None = None
|
|
|
|
|
|
def set_config_path(path: Path) -> None:
|
|
"""Set the current config path (used to derive data directory)."""
|
|
global _current_config_path
|
|
_current_config_path = path
|
|
|
|
|
|
def get_config_path() -> Path:
|
|
"""Get the configuration file path."""
|
|
if _current_config_path:
|
|
return _current_config_path
|
|
return Path.home() / ".nanobot" / "config.json"
|
|
|
|
|
|
def load_config(config_path: Path | None = None) -> Config:
|
|
"""
|
|
Load configuration from file or create default.
|
|
|
|
Args:
|
|
config_path: Optional path to config file. Uses default if not provided.
|
|
|
|
Returns:
|
|
Loaded configuration object.
|
|
"""
|
|
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:
|
|
logger.warning(f"Failed to load config from {path}: {e}")
|
|
logger.warning("Using default configuration.")
|
|
|
|
_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)
|
|
|
|
|
|
def save_config(config: Config, config_path: Path | None = None) -> None:
|
|
"""
|
|
Save configuration to file.
|
|
|
|
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)
|
|
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
json.dump(data, f, indent=2, ensure_ascii=False)
|
|
|
|
|
|
def resolve_config_env_vars(config: Config) -> Config:
|
|
"""Return a copy of *config* with ``${VAR}`` env-var references resolved.
|
|
|
|
Only string values are affected; other types pass through unchanged.
|
|
Raises :class:`ValueError` if a referenced variable is not set.
|
|
"""
|
|
data = config.model_dump(mode="json", by_alias=True)
|
|
data = _resolve_env_vars(data)
|
|
return Config.model_validate(data)
|
|
|
|
|
|
def _resolve_env_vars(obj: object) -> object:
|
|
"""Recursively resolve ``${VAR}`` patterns in string values."""
|
|
if isinstance(obj, str):
|
|
return re.sub(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}", _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."""
|
|
# 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
|