| 1 | """Configuration loading utilities.""" |
| 2 | |
| 3 | import json |
| 4 | import os |
| 5 | import re |
| 6 | from pathlib import Path |
| 7 | from typing import Any |
| 8 | |
| 9 | import pydantic |
| 10 | from loguru import logger |
| 11 | from pydantic import BaseModel |
| 12 | |
| 13 | from nanobot.config.schema import Config |
| 14 | |
| 15 | # Global variable to store current config path (for multi-instance support) |
| 16 | _current_config_path: Path | None = None |
| 17 | |
| 18 | |
| 19 | def set_config_path(path: Path) -> None: |
| 20 | """Set the current config path (used to derive data directory).""" |
| 21 | global _current_config_path |
| 22 | _current_config_path = path |
| 23 | |
| 24 | |
| 25 | def get_config_path() -> Path: |
| 26 | """Get the configuration file path.""" |
| 27 | if _current_config_path: |
| 28 | return _current_config_path |
| 29 | return Path.home() / ".nanobot" / "config.json" |
| 30 | |
| 31 | |
| 32 | def load_config(config_path: Path | None = None) -> Config: |
| 33 | """ |
| 34 | Load configuration from file or create default. |
| 35 | |
| 36 | Args: |
| 37 | config_path: Optional path to config file. Uses default if not provided. |
| 38 | |
| 39 | Returns: |
| 40 | Loaded configuration object. |
| 41 | """ |
| 42 | path = config_path or get_config_path() |
| 43 | |
| 44 | config = Config() |
| 45 | if path.exists(): |
| 46 | try: |
| 47 | with open(path, encoding="utf-8") as f: |
| 48 | data = json.load(f) |
| 49 | data = _migrate_config(data) |
| 50 | config = Config.model_validate(data) |
| 51 | except (json.JSONDecodeError, ValueError, pydantic.ValidationError) as e: |
| 52 | logger.warning(f"Failed to load config from {path}: {e}") |
| 53 | logger.warning("Using default configuration.") |
| 54 | |
| 55 | _apply_ssrf_whitelist(config) |
| 56 | return config |
| 57 | |
| 58 | |
| 59 | def _apply_ssrf_whitelist(config: Config) -> None: |
| 60 | """Apply SSRF whitelist from config to the network security module.""" |
| 61 | from nanobot.security.network import configure_ssrf_whitelist |
| 62 | |
| 63 | configure_ssrf_whitelist(config.tools.ssrf_whitelist) |
| 64 | |
| 65 | |
| 66 | def save_config(config: Config, config_path: Path | None = None) -> None: |
| 67 | """ |
| 68 | Save configuration to file. |
| 69 | |
| 70 | Args: |
| 71 | config: Configuration to save. |
| 72 | config_path: Optional path to save to. Uses default if not provided. |
| 73 | """ |
| 74 | path = config_path or get_config_path() |
| 75 | path.parent.mkdir(parents=True, exist_ok=True) |
| 76 | |
| 77 | data = config.model_dump(mode="json", by_alias=True) |
| 78 | |
| 79 | with open(path, "w", encoding="utf-8") as f: |
| 80 | json.dump(data, f, indent=2, ensure_ascii=False) |
| 81 | |
| 82 | |
| 83 | _ENV_REF_PATTERN = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}") |
| 84 | |
| 85 | |
| 86 | def resolve_config_env_vars(config: Config) -> Config: |
| 87 | """Return *config* with ``${VAR}`` env-var references resolved. |
| 88 | |
| 89 | Walks in place so fields declared with ``exclude=True`` (e.g. |
| 90 | ``DreamConfig.cron``) survive; returns the same instance when no |
| 91 | references are present. Raises ``ValueError`` if a referenced |
| 92 | variable is not set. |
| 93 | """ |
| 94 | return _resolve_in_place(config) |
| 95 | |
| 96 | |
| 97 | def _resolve_in_place(obj: Any) -> Any: |
| 98 | if isinstance(obj, str): |
| 99 | new = _ENV_REF_PATTERN.sub(_env_replace, obj) |
| 100 | return new if new != obj else obj |
| 101 | if isinstance(obj, BaseModel): |
| 102 | updates: dict[str, Any] = {} |
| 103 | for name in type(obj).model_fields: |
| 104 | old = getattr(obj, name) |
| 105 | new = _resolve_in_place(old) |
| 106 | if new is not old: |
| 107 | updates[name] = new |
| 108 | extras = obj.__pydantic_extra__ |
| 109 | new_extras: dict[str, Any] | None = None |
| 110 | if extras: |
| 111 | resolved = {k: _resolve_in_place(v) for k, v in extras.items()} |
| 112 | if any(resolved[k] is not extras[k] for k in extras): |
| 113 | new_extras = resolved |
| 114 | if not updates and new_extras is None: |
| 115 | return obj |
| 116 | copy = obj.model_copy(update=updates) if updates else obj.model_copy() |
| 117 | if new_extras is not None: |
| 118 | copy.__pydantic_extra__ = new_extras |
| 119 | return copy |
| 120 | if isinstance(obj, dict): |
| 121 | resolved = {k: _resolve_in_place(v) for k, v in obj.items()} |
| 122 | return resolved if any(resolved[k] is not obj[k] for k in obj) else obj |
| 123 | if isinstance(obj, list): |
| 124 | resolved = [_resolve_in_place(v) for v in obj] |
| 125 | return resolved if any(nv is not ov for nv, ov in zip(resolved, obj)) else obj |
| 126 | return obj |
| 127 | |
| 128 | |
| 129 | def _resolve_env_vars(obj: object) -> object: |
| 130 | """Recursively resolve ``${VAR}`` patterns in plain strings/dicts/lists.""" |
| 131 | if isinstance(obj, str): |
| 132 | return _ENV_REF_PATTERN.sub(_env_replace, obj) |
| 133 | if isinstance(obj, dict): |
| 134 | return {k: _resolve_env_vars(v) for k, v in obj.items()} |
| 135 | if isinstance(obj, list): |
| 136 | return [_resolve_env_vars(v) for v in obj] |
| 137 | return obj |
| 138 | |
| 139 | |
| 140 | def _env_replace(match: re.Match[str]) -> str: |
| 141 | name = match.group(1) |
| 142 | value = os.environ.get(name) |
| 143 | if value is None: |
| 144 | raise ValueError( |
| 145 | f"Environment variable '{name}' referenced in config is not set" |
| 146 | ) |
| 147 | return value |
| 148 | |
| 149 | |
| 150 | def _migrate_config(data: dict) -> dict: |
| 151 | """Migrate old config formats to current.""" |
| 152 | # Move tools.exec.restrictToWorkspace → tools.restrictToWorkspace |
| 153 | tools = data.get("tools", {}) |
| 154 | exec_cfg = tools.get("exec", {}) |
| 155 | if "restrictToWorkspace" in exec_cfg and "restrictToWorkspace" not in tools: |
| 156 | tools["restrictToWorkspace"] = exec_cfg.pop("restrictToWorkspace") |
| 157 | |
| 158 | # Move tools.myEnabled / tools.mySet → tools.my.{enable, allowSet}. |
| 159 | # The old flat keys shipped in the initial MyTool landing; wrapping them in a |
| 160 | # sub-config keeps `web` / `exec` / `my` symmetric and gives room to grow. |
| 161 | if "myEnabled" in tools or "mySet" in tools: |
| 162 | my_cfg = tools.setdefault("my", {}) |
| 163 | if "myEnabled" in tools and "enable" not in my_cfg: |
| 164 | my_cfg["enable"] = tools.pop("myEnabled") |
| 165 | else: |
| 166 | tools.pop("myEnabled", None) |
| 167 | if "mySet" in tools and "allowSet" not in my_cfg: |
| 168 | my_cfg["allowSet"] = tools.pop("mySet") |
| 169 | else: |
| 170 | tools.pop("mySet", None) |
| 171 | |
| 172 | # Move the legacy Memory VLM model out of the video-service section. |
| 173 | echo = ( |
| 174 | tools.get("echoGenerator") |
| 175 | or tools.get("echo_generator") |
| 176 | or tools.get("directorRemote") |
| 177 | or tools.get("director_remote") |
| 178 | or {} |
| 179 | ) |
| 180 | memory_review = tools.get("memoryReview") or tools.get("memory_review") |
| 181 | if memory_review is None: |
| 182 | memory_review = tools.setdefault("memoryReview", {}) |
| 183 | legacy_vlm_model = echo.pop("vlmModel", None) or echo.pop("vlm_model", None) |
| 184 | if legacy_vlm_model and "model" not in memory_review: |
| 185 | memory_review["model"] = legacy_vlm_model |
| 186 | |
| 187 | return data |
| 188 |