424 lines
17 KiB
Python
424 lines
17 KiB
Python
"""
|
|
Configuration manager for MaxClaw Agent
|
|
Handles persistent storage of provider configurations with encryption and validation
|
|
"""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Dict, Any, Optional, List, Tuple
|
|
import copy
|
|
from cryptography.fernet import Fernet
|
|
import base64
|
|
|
|
try:
|
|
import requests
|
|
except ImportError:
|
|
requests = None
|
|
|
|
|
|
class ConfigManager:
|
|
"""Manages MaxClaw Agent configuration with encryption and validation"""
|
|
|
|
def __init__(self, config_dir: Optional[str] = None):
|
|
"""
|
|
Initialize the config manager
|
|
|
|
Args:
|
|
config_dir: Directory to store config files. If None, uses ~/.maxclaw
|
|
"""
|
|
if config_dir is None:
|
|
self.config_dir = Path.home() / ".maxclaw"
|
|
else:
|
|
self.config_dir = Path(config_dir)
|
|
|
|
self.config_file = self.config_dir / "config.json"
|
|
self.key_file = self.config_dir / ".key"
|
|
|
|
# Ensure config directory exists with secure permissions
|
|
self._ensure_config_dir()
|
|
|
|
# Load or create encryption key
|
|
self._encryption_key = self._load_or_create_key()
|
|
self._cipher = Fernet(self._encryption_key)
|
|
|
|
# Load configuration
|
|
self.config = self._load_config()
|
|
|
|
def _ensure_config_dir(self):
|
|
"""Ensure the configuration directory exists with secure permissions"""
|
|
self.config_dir.mkdir(parents=True, exist_ok=True)
|
|
# Set directory permissions to owner-only (700)
|
|
os.chmod(self.config_dir, 0o700)
|
|
|
|
def _load_or_create_key(self) -> bytes:
|
|
"""Load existing encryption key or create a new one"""
|
|
if self.key_file.exists():
|
|
try:
|
|
key = self.key_file.read_bytes()
|
|
# Validate key format
|
|
Fernet(key) # This will raise an exception if invalid
|
|
# Set key file permissions to owner-only (600)
|
|
os.chmod(self.key_file, 0o600)
|
|
return key
|
|
except Exception:
|
|
# If key is invalid or corrupted, create a new one
|
|
pass
|
|
|
|
# Generate new key
|
|
key = Fernet.generate_key()
|
|
self.key_file.write_bytes(key)
|
|
# Set key file permissions to owner-only (600)
|
|
os.chmod(self.key_file, 0o600)
|
|
return key
|
|
|
|
def _encrypt_data(self, data: str) -> str:
|
|
"""Encrypt sensitive data"""
|
|
if not data:
|
|
return data
|
|
encrypted = self._cipher.encrypt(data.encode())
|
|
return base64.b64encode(encrypted).decode('utf-8')
|
|
|
|
def _decrypt_data(self, encrypted_data: str) -> str:
|
|
"""Decrypt sensitive data"""
|
|
if not encrypted_data:
|
|
return encrypted_data
|
|
try:
|
|
decoded = base64.b64decode(encrypted_data.encode())
|
|
decrypted = self._cipher.decrypt(decoded)
|
|
return decrypted.decode('utf-8')
|
|
except Exception:
|
|
# If decryption fails, return as-is (might be unencrypted legacy data)
|
|
return encrypted_data
|
|
|
|
def _load_config(self) -> Dict[str, Any]:
|
|
"""Load configuration from file"""
|
|
if self.config_file.exists():
|
|
try:
|
|
with open(self.config_file, 'r') as f:
|
|
config = json.load(f)
|
|
|
|
# Decrypt any encrypted fields in existing config
|
|
return self._decrypt_config(config)
|
|
except (json.JSONDecodeError, IOError) as e:
|
|
# Return default config if file is corrupted
|
|
print(f"Warning: Could not load config file ({e}), using defaults")
|
|
return self._get_default_config()
|
|
else:
|
|
# Return default config if file doesn't exist
|
|
return self._get_default_config()
|
|
|
|
def _decrypt_config(self, config: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Decrypt sensitive fields in configuration"""
|
|
if not isinstance(config, dict):
|
|
return config
|
|
|
|
# Create a deep copy to avoid modifying original
|
|
result = copy.deepcopy(config)
|
|
|
|
# Decrypt configurations
|
|
if "configurations" in result and isinstance(result["configurations"], dict):
|
|
for name, config_data in result["configurations"].items():
|
|
if isinstance(config_data, dict):
|
|
# Decrypt API keys for relevant providers
|
|
provider = config_data.get("provider", "").lower()
|
|
if provider in ["openai", "openrouter"] and "api_key" in config_data:
|
|
config_data["api_key"] = self._decrypt_data(config_data["api_key"])
|
|
|
|
return result
|
|
|
|
def _encrypt_config(self, config: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Encrypt sensitive fields in configuration before storage"""
|
|
if not isinstance(config, dict):
|
|
return config
|
|
|
|
# Create a deep copy to avoid modifying original
|
|
result = copy.deepcopy(config)
|
|
|
|
# Encrypt configurations
|
|
if "configurations" in result and isinstance(result["configurations"], dict):
|
|
for name, config_data in result["configurations"].items():
|
|
if isinstance(config_data, dict):
|
|
# Encrypt API keys for relevant providers
|
|
provider = config_data.get("provider", "").lower()
|
|
if provider in ["openai", "openrouter"] and "api_key" in config_data:
|
|
api_key = config_data["api_key"]
|
|
# Only encrypt if not already encrypted (avoid double encryption)
|
|
if api_key and not self._is_likely_encrypted(api_key):
|
|
config_data["api_key"] = self._encrypt_data(api_key)
|
|
|
|
return result
|
|
|
|
def _is_likely_encrypted(self, data: str) -> bool:
|
|
"""Heuristic to check if data appears to be encrypted"""
|
|
if not data or len(data) < 10:
|
|
return False
|
|
# Encrypted data with Fernet is base64 encoded and typically longer
|
|
try:
|
|
# Try to decode as base64
|
|
base64.b64decode(data.encode(), validate=True)
|
|
# If it decodes successfully and is reasonably long, likely encrypted
|
|
return len(data) >= 20
|
|
except Exception:
|
|
return False
|
|
|
|
def _get_default_config(self) -> Dict[str, Any]:
|
|
"""Get default configuration structure"""
|
|
return {
|
|
"configurations": {},
|
|
"active_configuration": None,
|
|
"version": "1.0"
|
|
}
|
|
|
|
def _save_config(self):
|
|
"""Save configuration to file with encryption of sensitive data"""
|
|
try:
|
|
# Encrypt sensitive data before saving
|
|
config_to_save = self._encrypt_config(self.config)
|
|
|
|
# Create a backup of the existing config
|
|
if self.config_file.exists():
|
|
backup_file = self.config_file.with_suffix('.json.bak')
|
|
backup_file.write_text(self.config_file.read_text())
|
|
# Set backup file permissions to owner-only (600)
|
|
os.chmod(backup_file, 0o600)
|
|
|
|
# Write the encrypted config
|
|
with open(self.config_file, 'w') as f:
|
|
json.dump(config_to_save, f, indent=2, sort_keys=True)
|
|
|
|
# Set config file permissions to owner-only (600)
|
|
os.chmod(self.config_file, 0o600)
|
|
except IOError as e:
|
|
raise Exception(f"Failed to save configuration: {e}")
|
|
|
|
def validate_configuration(self, config: Dict[str, Any]) -> Tuple[bool, List[str]]:
|
|
"""
|
|
Validate a configuration dictionary
|
|
|
|
Returns:
|
|
Tuple of (is_valid, list_of_error_messages)
|
|
"""
|
|
errors = []
|
|
|
|
# Check if config is a dictionary
|
|
if not isinstance(config, dict):
|
|
errors.append("Configuration must be a dictionary")
|
|
return False, errors
|
|
|
|
# Validate provider
|
|
provider = config.get("provider", "").lower()
|
|
valid_providers = ["openai", "ollama", "openrouter"]
|
|
|
|
if not provider:
|
|
errors.append("Provider is required")
|
|
elif provider not in valid_providers:
|
|
errors.append(f"Unsupported provider: {provider}. Valid options: {', '.join(valid_providers)}")
|
|
|
|
# Validate model name
|
|
model_name = config.get("model_name", "").strip()
|
|
if not model_name:
|
|
errors.append("Model name is required and cannot be empty")
|
|
elif len(model_name) > 100:
|
|
errors.append("Model name is too long (maximum 100 characters)")
|
|
|
|
# Provider-specific validation
|
|
if provider == "openai":
|
|
api_key = config.get("api_key", "").strip()
|
|
if not api_key:
|
|
errors.append("API key is required for OpenAI provider")
|
|
elif not api_key.startswith("sk-"):
|
|
errors.append("OpenAI API key should start with 'sk-'")
|
|
elif len(api_key) < 20:
|
|
errors.append("OpenAI API key appears too short")
|
|
|
|
elif provider == "ollama":
|
|
base_url = config.get("base_url", "").strip()
|
|
if not base_url:
|
|
errors.append("Base URL is required for Ollama provider")
|
|
else:
|
|
# Normalize URL for validation
|
|
if not base_url.startswith(("http://", "https://")):
|
|
errors.append("Ollama base URL must start with http:// or https://")
|
|
else:
|
|
# Remove trailing slash for consistency
|
|
if base_url.endswith("/"):
|
|
config["base_url"] = base_url.rstrip("/")
|
|
|
|
elif provider == "openrouter":
|
|
api_key = config.get("api_key", "").strip()
|
|
base_url = config.get("base_url", "").strip()
|
|
|
|
if not api_key:
|
|
errors.append("API key is required for OpenRouter provider")
|
|
elif len(api_key) < 10:
|
|
errors.append("OpenRouter API key appears too short")
|
|
|
|
if not base_url:
|
|
errors.append("Base URL is required for OpenRouter provider")
|
|
elif not base_url.startswith(("http://", "https://")):
|
|
errors.append("OpenRouter base URL must start with http:// or https://")
|
|
else:
|
|
# Remove trailing slash for consistency
|
|
if base_url.endswith("/"):
|
|
config["base_url"] = base_url.rstrip("/")
|
|
|
|
return len(errors) == 0, errors
|
|
|
|
def get_configurations(self) -> Dict[str, Any]:
|
|
"""Get all configurations"""
|
|
return self.config.get("configurations", {})
|
|
|
|
def get_active_configuration(self) -> Optional[str]:
|
|
"""Get the name of the active configuration"""
|
|
return self.config.get("active_configuration")
|
|
|
|
def get_configuration(self, name: str) -> Optional[Dict[str, Any]]:
|
|
"""Get a specific configuration by name"""
|
|
configurations = self.get_configurations()
|
|
config = configurations.get(name)
|
|
# Return a copy to prevent accidental modification of internal data
|
|
return copy.deepcopy(config) if config else None
|
|
|
|
def add_configuration(self, name: str, config: Dict[str, Any]):
|
|
"""
|
|
Add or update a configuration with validation
|
|
|
|
Args:
|
|
name: Configuration name
|
|
config: Configuration dictionary with provider, model, etc.
|
|
|
|
Raises:
|
|
ValueError: If configuration is invalid
|
|
"""
|
|
if not name or not isinstance(name, str):
|
|
raise ValueError("Configuration name must be a non-empty string")
|
|
|
|
# Validate the configuration
|
|
is_valid, errors = self.validate_configuration(config)
|
|
if not is_valid:
|
|
raise ValueError(f"Invalid configuration: {', '.join(errors)}")
|
|
|
|
if "configurations" not in self.config:
|
|
self.config["configurations"] = {}
|
|
|
|
# Deep copy to avoid reference issues
|
|
self.config["configurations"][name] = copy.deepcopy(config)
|
|
self._save_config()
|
|
|
|
def remove_configuration(self, name: str) -> bool:
|
|
"""
|
|
Remove a configuration
|
|
|
|
Args:
|
|
name: Configuration name to remove
|
|
|
|
Returns:
|
|
True if removed, False if not found
|
|
"""
|
|
if "configurations" in self.config and name in self.config["configurations"]:
|
|
del self.config["configurations"][name]
|
|
|
|
# If this was the active configuration, unset it
|
|
if self.config.get("active_configuration") == name:
|
|
self.config["active_configuration"] = None
|
|
|
|
self._save_config()
|
|
return True
|
|
return False
|
|
|
|
def set_active_configuration(self, name: str) -> bool:
|
|
"""
|
|
Set the active configuration
|
|
|
|
Args:
|
|
name: Configuration name to set as active
|
|
|
|
Returns:
|
|
True if successful, False if configuration not found
|
|
"""
|
|
if name in self.get_configurations():
|
|
self.config["active_configuration"] = name
|
|
self._save_config()
|
|
return True
|
|
return False
|
|
|
|
def get_active_config_details(self) -> Optional[Dict[str, Any]]:
|
|
"""Get the details of the active configuration"""
|
|
active_name = self.get_active_configuration()
|
|
if active_name:
|
|
return self.get_configuration(active_name)
|
|
return None
|
|
|
|
def list_configuration_names(self) -> List[str]:
|
|
"""Get a list of all configuration names"""
|
|
return list(self.get_configurations().keys())
|
|
|
|
def configuration_exists(self, name: str) -> bool:
|
|
"""Check if a configuration exists"""
|
|
return name in self.get_configurations()
|
|
|
|
def test_configuration_connection(self, name: str) -> Tuple[bool, str]:
|
|
"""
|
|
Test connection to a configured provider
|
|
|
|
Args:
|
|
name: Configuration name to test
|
|
|
|
Returns:
|
|
Tuple of (success, message)
|
|
"""
|
|
config = self.get_configuration(name)
|
|
if not config:
|
|
return False, f"Configuration '{name}' not found"
|
|
|
|
provider = config.get("provider", "").lower()
|
|
|
|
# Check if requests is available
|
|
if requests is None:
|
|
return False, "The 'requests' library is not available. Please install it to test connections."
|
|
|
|
try:
|
|
if provider == "ollama":
|
|
base_url = config.get("base_url", "http://localhost:11434")
|
|
# Ensure no trailing slash
|
|
if base_url.endswith("/"):
|
|
base_url = base_url.rstrip("/")
|
|
response = requests.get(f"{base_url}/api/tags", timeout=5)
|
|
if response.status_code == 200:
|
|
models = response.json().get("models", [])
|
|
model_names = [m.get("name", "unknown") for m in models]
|
|
return True, f"Connected successfully. Available models: {', '.join(model_names[:5])}{'...' if len(model_names) > 5 else ''}"
|
|
else:
|
|
return False, f"Ollama server returned status {response.status_code}"
|
|
|
|
elif provider == "openai":
|
|
# For OpenAI, we can't easily test without making an actual API call
|
|
# which might incur costs. Just validate the key format.
|
|
api_key = config.get("api_key", "")
|
|
if api_key and api_key.startswith("sk-") and len(api_key) > 20:
|
|
return True, "API key format appears valid (no live test performed to avoid costs)"
|
|
else:
|
|
return False, "API key format appears invalid"
|
|
|
|
elif provider == "openrouter":
|
|
# Similar to OpenAI, just validate format to avoid costs
|
|
api_key = config.get("api_key", "")
|
|
base_url = config.get("base_url", "https://openrouter.ai/api/v1")
|
|
if api_key and len(api_key) > 10:
|
|
return True, "API key format appears valid (no live test performed to avoid costs)"
|
|
else:
|
|
return False, "API key format appears invalid"
|
|
|
|
else:
|
|
return False, f"Connection testing not implemented for provider: {provider}"
|
|
|
|
except requests.exceptions.Timeout:
|
|
return False, "Connection timeout - please check the URL and network connectivity"
|
|
except requests.exceptions.ConnectionError:
|
|
return False, "Connection failed - please check the URL and ensure the service is running"
|
|
except Exception as e:
|
|
return False, f"Connection test failed: {str(e)}"
|
|
|
|
|
|
# Global config manager instance
|
|
config_manager = ConfigManager() |