| """
|
| API Key management for the Agentic AI System.
|
| Loads API keys from configuration file with fallback to defaults.
|
| """
|
|
|
| import json
|
| import os
|
| import secrets
|
| from pathlib import Path
|
| from typing import Dict, Any, List
|
| from src.utils.logging_config import logger
|
|
|
|
|
| class APIKeyManager:
|
| """Manages API keys and authentication configuration."""
|
|
|
| def __init__(self, config_file: str = "config/api_keys.json"):
|
| """Initialize API key manager."""
|
| self.config_file = Path(config_file)
|
| self.api_keys = {}
|
| self.settings = {}
|
| self.load_config()
|
|
|
| def load_config(self) -> None:
|
| """Load API keys from configuration file."""
|
| try:
|
| if self.config_file.exists():
|
| with open(self.config_file, 'r') as f:
|
| config = json.load(f)
|
|
|
| self.api_keys = config.get('api_keys', {})
|
| self.settings = config.get('settings', {})
|
|
|
|
|
| self.api_keys = {
|
| key: value for key, value in self.api_keys.items()
|
| if value.get('enabled', True)
|
| }
|
|
|
| logger.info(f"Loaded {len(self.api_keys)} API keys from {self.config_file}")
|
| else:
|
| logger.warning(f"API keys file not found: {self.config_file}")
|
| self._load_default_keys()
|
|
|
| except Exception as e:
|
| logger.error(f"Failed to load API keys: {e}")
|
| self._load_default_keys()
|
|
|
| def _load_default_keys(self) -> None:
|
| """No config file found: generate ONE random read/write key and persist it.
|
|
|
| Security: never ship static, published tokens. A key can also be supplied
|
| out-of-band via the ADMIN_API_KEY env var (useful for ephemeral/HF-style
|
| deploys where writing a config file is undesirable). Admin permission is
|
| never granted by default — add an admin key explicitly with manage_api_keys.
|
| """
|
| env_key = os.getenv("ADMIN_API_KEY")
|
| generated = env_key or secrets.token_urlsafe(32)
|
| self.api_keys = {
|
| generated: {
|
| "name": "Auto-generated key",
|
| "permissions": ["read", "write"],
|
| "description": "Generated at first start (no config/api_keys.json found)",
|
| "rate_limit": 100,
|
| "enabled": True,
|
| }
|
| }
|
| self.settings = {
|
| "default_rate_limit": 100,
|
| "token_expiry_days": 30,
|
| "require_https": True,
|
| }
|
| if env_key:
|
| logger.info("Loaded API key from ADMIN_API_KEY env var")
|
| else:
|
|
|
| logger.warning(
|
| "No config/api_keys.json found. Generated a new API key "
|
| "(shown once): %s", generated
|
| )
|
| try:
|
| self.save_config()
|
| except Exception as e:
|
| logger.error("Could not persist generated API key: %s", e)
|
|
|
| def get_api_keys(self) -> Dict[str, Any]:
|
| """Get all enabled API keys."""
|
| return self.api_keys.copy()
|
|
|
| def validate_key(self, key: str) -> bool:
|
| """Check if API key is valid."""
|
| return key in self.api_keys and self.api_keys[key].get('enabled', True)
|
|
|
| def get_key_info(self, key: str) -> Dict[str, Any]:
|
| """Get information about a specific API key."""
|
| return self.api_keys.get(key, {})
|
|
|
| def has_permission(self, key: str, permission: str) -> bool:
|
| """Check if API key has specific permission."""
|
| if key not in self.api_keys:
|
| return False
|
|
|
| permissions = self.api_keys[key].get('permissions', [])
|
| return permission in permissions
|
|
|
| def get_rate_limit(self, key: str) -> int:
|
| """Get rate limit for API key."""
|
| if key not in self.api_keys:
|
| return self.settings.get('default_rate_limit', 100)
|
|
|
| return self.api_keys[key].get('rate_limit', self.settings.get('default_rate_limit', 100))
|
|
|
| def add_key(self, key: str, config: Dict[str, Any]) -> bool:
|
| """Add new API key."""
|
| try:
|
| self.api_keys[key] = config
|
| self.save_config()
|
| logger.info(f"Added new API key: {config.get('name', 'Unknown')}")
|
| return True
|
| except Exception as e:
|
| logger.error(f"Failed to add API key: {e}")
|
| return False
|
|
|
| def remove_key(self, key: str) -> bool:
|
| """Remove API key."""
|
| try:
|
| if key in self.api_keys:
|
| removed = self.api_keys.pop(key)
|
| self.save_config()
|
| logger.info(f"Removed API key: {removed.get('name', 'Unknown')}")
|
| return True
|
| return False
|
| except Exception as e:
|
| logger.error(f"Failed to remove API key: {e}")
|
| return False
|
|
|
| def disable_key(self, key: str) -> bool:
|
| """Disable API key without removing it."""
|
| try:
|
| if key in self.api_keys:
|
| self.api_keys[key]['enabled'] = False
|
| self.save_config()
|
| logger.info(f"Disabled API key: {key}")
|
| return True
|
| return False
|
| except Exception as e:
|
| logger.error(f"Failed to disable API key: {e}")
|
| return False
|
|
|
| def save_config(self) -> bool:
|
| """Save current configuration to file."""
|
| try:
|
|
|
| self.config_file.parent.mkdir(parents=True, exist_ok=True)
|
|
|
| config = {
|
| "api_keys": self.api_keys,
|
| "settings": self.settings
|
| }
|
|
|
| with open(self.config_file, 'w') as f:
|
| json.dump(config, f, indent=2)
|
|
|
| logger.info(f"Saved API keys configuration to {self.config_file}")
|
| return True
|
|
|
| except Exception as e:
|
| logger.error(f"Failed to save API keys: {e}")
|
| return False
|
|
|
| def list_keys(self) -> List[Dict[str, Any]]:
|
| """List all API keys with their info (excluding the actual key)."""
|
| keys_info = []
|
| for key, info in self.api_keys.items():
|
| key_info = {
|
| "key_id": key[:8] + "..." + key[-4:],
|
| "name": info.get('name', 'Unknown'),
|
| "permissions": info.get('permissions', []),
|
| "rate_limit": info.get('rate_limit', 100),
|
| "enabled": info.get('enabled', True),
|
| "description": info.get('description', '')
|
| }
|
| keys_info.append(key_info)
|
|
|
| return keys_info
|
|
|
| def reload(self) -> None:
|
| """Reload configuration from file."""
|
| self.load_config()
|
|
|
|
|
|
|
| api_key_manager = APIKeyManager()
|
|
|
