# -*- coding: utf-8 -*- """ Verschlüsselter Secrets-Provider. Speichert Secrets in einer AES-256 verschlüsselten Datei. """ from __future__ import annotations import base64 import hashlib import json import os from datetime import datetime from pathlib import Path from typing import Any, TypeVar from trixy_core.config.secrets.provider import SecretsProvider, SecretValue T = TypeVar("T") class EncryptedSecretsProvider(SecretsProvider): """ Provider für verschlüsselte Secrets. Verwendet AES-256-GCM für die Verschlüsselung. Secrets werden in einer JSON-Datei gespeichert. Example: # Erstellen mit Passwort provider = EncryptedSecretsProvider( path="secrets.enc", password="my-secret-password" ) # Secret speichern provider.set("database_password", "supersecret123") # Secret laden password = provider.get("database_password") """ # AES-256-GCM Parameter KEY_SIZE = 32 # 256 bits NONCE_SIZE = 12 TAG_SIZE = 16 SALT_SIZE = 16 ITERATIONS = 100000 def __init__( self, path: str | Path, password: str | None = None, password_env_var: str = "TRIXY_SECRETS_PASSWORD", auto_save: bool = True, ) -> None: """ Initialisiert den verschlüsselten Provider. Args: path: Pfad zur verschlüsselten Datei. password: Passwort für die Verschlüsselung. password_env_var: Umgebungsvariable für das Passwort. auto_save: Ob Änderungen automatisch gespeichert werden. """ super().__init__("EncryptedSecretsProvider") self._path = Path(path) self._auto_save = auto_save self._secrets: dict[str, Any] = {} self._dirty = False # Passwort ermitteln self._password = password or os.environ.get(password_env_var) if not self._password: raise ValueError( f"Passwort erforderlich. Setze '{password_env_var}' oder übergebe 'password'" ) # Schlüssel ableiten self._salt: bytes | None = None self._key: bytes | None = None # Datei laden falls vorhanden if self._path.exists(): self._load() def _derive_key(self, salt: bytes) -> bytes: """Leitet den Verschlüsselungsschlüssel vom Passwort ab.""" return hashlib.pbkdf2_hmac( "sha256", self._password.encode("utf-8"), salt, self.ITERATIONS, dklen=self.KEY_SIZE, ) def _encrypt(self, data: bytes) -> bytes: """Verschlüsselt Daten mit AES-256-GCM.""" try: from cryptography.hazmat.primitives.ciphers.aead import AESGCM except ImportError: raise ImportError( "cryptography ist erforderlich für EncryptedSecretsProvider" ) # Neuen Salt generieren salt = os.urandom(self.SALT_SIZE) key = self._derive_key(salt) # Verschlüsseln nonce = os.urandom(self.NONCE_SIZE) aesgcm = AESGCM(key) ciphertext = aesgcm.encrypt(nonce, data, None) # Format: salt + nonce + ciphertext (enthält Tag) return salt + nonce + ciphertext def _decrypt(self, encrypted: bytes) -> bytes: """Entschlüsselt Daten mit AES-256-GCM.""" try: from cryptography.hazmat.primitives.ciphers.aead import AESGCM except ImportError: raise ImportError( "cryptography ist erforderlich für EncryptedSecretsProvider" ) # Daten aufteilen salt = encrypted[:self.SALT_SIZE] nonce = encrypted[self.SALT_SIZE:self.SALT_SIZE + self.NONCE_SIZE] ciphertext = encrypted[self.SALT_SIZE + self.NONCE_SIZE:] # Schlüssel ableiten und entschlüsseln key = self._derive_key(salt) aesgcm = AESGCM(key) try: return aesgcm.decrypt(nonce, ciphertext, None) except Exception as e: raise ValueError(f"Entschlüsselung fehlgeschlagen: {e}") def _load(self) -> None: """Lädt und entschlüsselt die Secrets-Datei.""" with open(self._path, "rb") as f: encrypted = f.read() decrypted = self._decrypt(encrypted) self._secrets = json.loads(decrypted.decode("utf-8")) self._dirty = False def _save(self) -> None: """Verschlüsselt und speichert die Secrets.""" data = json.dumps(self._secrets).encode("utf-8") encrypted = self._encrypt(data) # Verzeichnis erstellen falls nötig self._path.parent.mkdir(parents=True, exist_ok=True) with open(self._path, "wb") as f: f.write(encrypted) self._dirty = False def get(self, key: str, default: T | None = None) -> T | None: """ Holt einen Secret-Wert. Args: key: Schlüssel. default: Standardwert. Returns: Secret-Wert oder Default. """ return self._secrets.get(key, default) def get_secret(self, key: str) -> SecretValue | None: """ Holt ein Secret mit Metadaten. Args: key: Schlüssel. Returns: SecretValue oder None. """ if key not in self._secrets: return None return SecretValue( key=key, value=self._secrets[key], source=str(self._path), ) def set(self, key: str, value: Any) -> bool: """ Speichert einen Secret-Wert. Args: key: Schlüssel. value: Wert. Returns: True wenn erfolgreich. """ self._secrets[key] = value self._dirty = True if self._auto_save: self._save() return True def delete(self, key: str) -> bool: """ Löscht ein Secret. Args: key: Schlüssel. Returns: True wenn vorhanden und gelöscht. """ if key in self._secrets: del self._secrets[key] self._dirty = True if self._auto_save: self._save() return True return False def exists(self, key: str) -> bool: """Prüft, ob ein Secret existiert.""" return key in self._secrets def list_keys(self) -> list[str]: """Listet alle Secret-Schlüssel.""" return list(self._secrets.keys()) def save(self) -> None: """Speichert Änderungen manuell.""" if self._dirty: self._save() def rotate_password(self, new_password: str) -> None: """ Rotiert das Passwort. Args: new_password: Neues Passwort. """ self._password = new_password self._save() def export_json(self) -> str: """ Exportiert Secrets als JSON (nur für Debug/Migration). Returns: JSON-String. """ return json.dumps(self._secrets, indent=2) def import_json(self, json_str: str) -> None: """ Importiert Secrets aus JSON. Args: json_str: JSON-String mit Secrets. """ imported = json.loads(json_str) self._secrets.update(imported) self._dirty = True if self._auto_save: self._save() @classmethod def create( cls, path: str | Path, password: str, initial_secrets: dict[str, Any] | None = None, ) -> "EncryptedSecretsProvider": """ Erstellt eine neue verschlüsselte Secrets-Datei. Args: path: Pfad zur Datei. password: Passwort. initial_secrets: Optionale initiale Secrets. Returns: EncryptedSecretsProvider-Instanz. """ provider = cls(path=path, password=password, auto_save=False) if initial_secrets: provider._secrets = initial_secrets provider._save() return provider