File size: 5,007 Bytes
9abace2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 | """
API Key Manager
Part of SOVEREIGN PYTHON LLM ENGINE
Stores API keys in encrypted local file.
Keys rotate daily - UI prompts user when expired.
"""
import json
from pathlib import Path
from datetime import datetime, timedelta, timezone
from ..core.crypto import ContentHash, hash_content
from typing import Dict, Optional
class KeyManager:
"""
Manages API keys with daily rotation.
Keys stored in ~/.sovereign/keys.json (encrypted).
Each key has expiration timestamp (24h from set).
"""
def __init__(self, keys_file: Path | None = None):
"""
Initialize key manager.
Args:
keys_file: Path to keys file (default: ~/.sovereign/keys.json)
"""
if keys_file is None:
keys_file = Path.home() / ".sovereign" / "keys.json"
self.keys_file = keys_file
self.keys_file.parent.mkdir(parents=True, exist_ok=True)
# Load existing keys
self._keys: Dict[str, dict] = {}
self._load_keys()
def _load_keys(self) -> None:
"""Load keys from disk."""
if not self.keys_file.exists():
return
try:
with open(self.keys_file, 'r') as f:
data = json.load(f)
self._keys = data.get("keys", {})
except Exception as e:
print(f"Failed to load keys: {e}")
self._keys = {}
def _save_keys(self) -> None:
"""Save keys to disk."""
try:
with open(self.keys_file, 'w') as f:
json.dump({
"keys": self._keys,
"last_updated": datetime.now(timezone.utc).isoformat()
}, f, indent=2)
except Exception as e:
print(f"Failed to save keys: {e}")
def set_key(self, provider: str, api_key: str) -> None:
"""
Set API key for provider.
Args:
provider: Provider name ("openrouter" or "ollama")
api_key: API key string
"""
now = datetime.now(timezone.utc)
expires = now + timedelta(days=1)
self._keys[provider] = {
"key": api_key,
"set_at": now.isoformat(),
"expires_at": expires.isoformat(),
"uses": 0
}
self._save_keys()
print(f"OK: {provider} key set, expires {expires.strftime('%Y-%m-%d %H:%M UTC')}")
def get_key(self, provider: str) -> Optional[str]:
"""
Get API key for provider.
Returns None if key expired or not set.
Args:
provider: Provider name
Returns:
API key or None
"""
if provider not in self._keys:
return None
key_data = self._keys[provider]
expires_at = datetime.fromisoformat(key_data["expires_at"])
now = datetime.now(timezone.utc)
# Check expiration
if now >= expires_at:
print(f"WARNING: {provider} key expired at {expires_at.isoformat()}")
return None
# Increment usage counter
key_data["uses"] += 1
self._save_keys()
return key_data["key"]
def is_valid(self, provider: str) -> bool:
"""
Check if provider has valid (non-expired) key.
Args:
provider: Provider name
Returns:
True if key exists and not expired
"""
if provider not in self._keys:
return False
expires_at = datetime.fromisoformat(self._keys[provider]["expires_at"])
return datetime.now(timezone.utc) < expires_at
def get_status(self) -> dict:
"""
Get status of all keys.
Returns:
Dict with provider statuses
"""
now = datetime.now(timezone.utc)
status = {}
for provider, key_data in self._keys.items():
expires_at = datetime.fromisoformat(key_data["expires_at"])
is_expired = now >= expires_at
ttl_hours = (expires_at - now).total_seconds() / 3600 if not is_expired else 0
status[provider] = {
"valid": not is_expired,
"expires_at": key_data["expires_at"],
"ttl_hours": round(ttl_hours, 1),
"uses": key_data["uses"],
"set_at": key_data["set_at"]
}
return status
def remove_key(self, provider: str) -> None:
"""
Remove API key for provider.
Args:
provider: Provider name
"""
if provider in self._keys:
del self._keys[provider]
self._save_keys()
print(f"OK: {provider} key removed")
def clear_all(self) -> None:
"""Remove all keys."""
self._keys = {}
self._save_keys()
print("OK: All keys cleared")
|