Trixy7/trixy_core/config/secrets/encrypted.py

# -*- 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