Coverage for src / osiris_cli / key_manager.py: 0%

1#!/usr/bin/env python3 

2""" 

3Secure API Key Manager for Osiris CLI 

4Stores API keys encrypted in user's home directory 

5""" 

6 

7import json 

8from pathlib import Path 

9from typing import Dict, Optional 

10from cryptography.fernet import Fernet 

11import base64 

12import hashlib 

13from rich.console import Console 

14 

15console = Console() 

16 

17 

18class SecureKeyManager: 

19 """Manage API keys securely with encryption""" 

20 

21 def __init__(self): 

22 self.keys_dir = Path.home() / ".osiris" / "keys" 

23 self.keys_dir.mkdir(parents=True, exist_ok=True) 

24 self.keys_file = self.keys_dir / "api_keys.enc" 

25 self.cipher = self._get_cipher() 

26 

27 def _get_cipher(self) -> Fernet: 

28 """Get or create encryption cipher""" 

29 # Use machine-specific key (derived from hostname + user) 

30 import socket 

31 import getpass 

32 

33 machine_id = f"{socket.gethostname()}-{getpass.getuser()}" 

34 key_material = hashlib.sha256(machine_id.encode()).digest() 

35 key = base64.urlsafe_b64encode(key_material) 

36 

37 return Fernet(key) 

38 

39 def save_key(self, provider: str, api_key: str): 

40 """Save API key securely""" 

41 # Load existing keys 

42 keys = self.load_all_keys() 

43 

44 # Add/update key 

45 keys[provider] = api_key 

46 

47 # Encrypt and save 

48 try: 

49 encrypted_data = self.cipher.encrypt(json.dumps(keys).encode()) 

50 self.keys_file.write_bytes(encrypted_data) 

51 console.print(f"[green]✓[/green] API key saved securely for {provider}") 

52 except Exception as e: 

53 console.print(f"[red]Error saving key: {e}[/red]") 

54 

55 def load_key(self, provider: str) -> Optional[str]: 

56 """Load API key for provider""" 

57 keys = self.load_all_keys() 

58 return keys.get(provider) 

59 

60 def load_all_keys(self) -> Dict[str, str]: 

61 """Load all saved API keys""" 

62 if not self.keys_file.exists(): 

63 return {} 

64 

65 try: 

66 encrypted_data = self.keys_file.read_bytes() 

67 decrypted_data = self.cipher.decrypt(encrypted_data) 

68 return json.loads(decrypted_data.decode()) 

69 except Exception: 

70 # If decryption fails, return empty dict 

71 return {} 

72 

73 def has_key(self, provider: str) -> bool: 

74 """Check if API key exists for provider""" 

75 return provider in self.load_all_keys() 

76 

77 def delete_key(self, provider: str): 

78 """Delete API key for provider""" 

79 keys = self.load_all_keys() 

80 if provider in keys: 

81 del keys[provider] 

82 

83 # Save updated keys 

84 try: 

85 encrypted_data = self.cipher.encrypt(json.dumps(keys).encode()) 

86 self.keys_file.write_bytes(encrypted_data) 

87 console.print(f"[green]✓[/green] API key deleted for {provider}") 

88 except Exception as e: 

89 console.print(f"[red]Error deleting key: {e}[/red]") 

90 

91 def list_providers_with_keys(self) -> list: 

92 """List all providers with saved keys""" 

93 return list(self.load_all_keys().keys()) 

94 

95 def mask_key(self, api_key: str) -> str: 

96 """Mask API key for display""" 

97 if len(api_key) <= 8: 

98 return "***" 

99 return f"{api_key[:4]}...{api_key[-4:]}" 

100 

101 

102# Global key manager instance 

103key_manager = SecureKeyManager()