Trixy7/trixy_core/config/secrets/provider.py

# -*- coding: utf-8 -*-
"""
Abstrakte Basisklasse für Secrets-Provider.

Definiert die Schnittstelle für alle Secrets-Provider
Implementierungen.
"""

from __future__ import annotations

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Generic, TypeVar


T = TypeVar("T")


@dataclass
class SecretValue(Generic[T]):
    """
    Wrapper für einen Secret-Wert.

    Bietet Metadaten und sichere Handhabung für Secrets.
    """

    key: str
    """Schlüssel/Name des Secrets."""

    value: T
    """Der eigentliche Secret-Wert."""

    source: str = ""
    """Quelle des Secrets (z.B. 'env', 'file')."""

    created_at: datetime = field(default_factory=datetime.now)
    """Zeitpunkt des Ladens."""

    expires_at: datetime | None = None
    """Optionaler Ablaufzeitpunkt."""

    metadata: dict[str, Any] = field(default_factory=dict)
    """Zusätzliche Metadaten."""

    @property
    def is_expired(self) -> bool:
        """Prüft, ob das Secret abgelaufen ist."""
        if self.expires_at is None:
            return False
        return datetime.now() > self.expires_at

    def __repr__(self) -> str:
        """Sichere Repräsentation ohne Wert."""
        return f"SecretValue(key={self.key!r}, source={self.source!r})"

    def __str__(self) -> str:
        """Sichere String-Darstellung."""
        return f"<Secret: {self.key}>"


class SecretsProvider(ABC):
    """
    Abstrakte Basisklasse für Secrets-Provider.

    Definiert die Schnittstelle zum Laden, Speichern und
    Verwalten von Secrets.

    Implementierungen können verschiedene Backends nutzen:
    - Umgebungsvariablen
    - Verschlüsselte Dateien
    - Cloud Secret Manager (AWS, Azure, GCP)
    - HashiCorp Vault
    """

    def __init__(self, name: str = "") -> None:
        """
        Initialisiert den Provider.

        Args:
            name: Name des Providers für Logging.
        """
        self._name = name or self.__class__.__name__
        self._cache: dict[str, SecretValue] = {}
        self._cache_enabled = True

    @property
    def name(self) -> str:
        """Gibt den Provider-Namen zurück."""
        return self._name

    @property
    def cache_enabled(self) -> bool:
        """Ob Caching aktiviert ist."""
        return self._cache_enabled

    @cache_enabled.setter
    def cache_enabled(self, value: bool) -> None:
        """Aktiviert/Deaktiviert Caching."""
        self._cache_enabled = value
        if not value:
            self._cache.clear()

    @abstractmethod
    def get(self, key: str, default: T | None = None) -> T | None:
        """
        Holt einen Secret-Wert.

        Args:
            key: Schlüssel des Secrets.
            default: Standardwert falls nicht gefunden.

        Returns:
            Secret-Wert oder Default.
        """
        pass

    @abstractmethod
    def get_secret(self, key: str) -> SecretValue | None:
        """
        Holt ein Secret mit Metadaten.

        Args:
            key: Schlüssel des Secrets.

        Returns:
            SecretValue oder None.
        """
        pass

    def get_required(self, key: str) -> Any:
        """
        Holt einen erforderlichen Secret-Wert.

        Args:
            key: Schlüssel des Secrets.

        Returns:
            Secret-Wert.

        Raises:
            KeyError: Wenn das Secret nicht gefunden wird.
        """
        value = self.get(key)
        if value is None:
            raise KeyError(f"Erforderliches Secret '{key}' nicht gefunden")
        return value

    def set(self, key: str, value: Any) -> bool:
        """
        Speichert einen Secret-Wert.

        Args:
            key: Schlüssel.
            value: Wert.

        Returns:
            True wenn erfolgreich.

        Note:
            Nicht alle Provider unterstützen Schreibzugriff.
        """
        raise NotImplementedError(
            f"{self._name} unterstützt kein Speichern von Secrets"
        )

    def delete(self, key: str) -> bool:
        """
        Löscht ein Secret.

        Args:
            key: Schlüssel.

        Returns:
            True wenn erfolgreich.
        """
        raise NotImplementedError(
            f"{self._name} unterstützt kein Löschen von Secrets"
        )

    def exists(self, key: str) -> bool:
        """
        Prüft, ob ein Secret existiert.

        Args:
            key: Schlüssel.

        Returns:
            True wenn vorhanden.
        """
        return self.get(key) is not None

    def list_keys(self) -> list[str]:
        """
        Listet alle verfügbaren Secret-Schlüssel.

        Returns:
            Liste der Schlüssel.
        """
        raise NotImplementedError(
            f"{self._name} unterstützt kein Auflisten von Secrets"
        )

    def clear_cache(self) -> None:
        """Löscht den Secret-Cache."""
        self._cache.clear()

    def refresh(self, key: str | None = None) -> None:
        """
        Aktualisiert gecachte Secrets.

        Args:
            key: Optionaler spezifischer Schlüssel.
        """
        if key:
            self._cache.pop(key, None)
        else:
            self._cache.clear()


class ChainedSecretsProvider(SecretsProvider):
    """
    Provider der mehrere Provider verkettet.

    Sucht Secrets in der Reihenfolge der registrierten Provider
    und gibt den ersten Treffer zurück.
    """

    def __init__(
        self,
        providers: list[SecretsProvider] | None = None,
        name: str = "ChainedSecretsProvider",
    ) -> None:
        """
        Initialisiert den verketteten Provider.

        Args:
            providers: Liste von Providern.
            name: Name des Providers.
        """
        super().__init__(name)
        self._providers = providers or []

    def add_provider(self, provider: SecretsProvider) -> "ChainedSecretsProvider":
        """
        Fügt einen Provider zur Kette hinzu.

        Args:
            provider: Der hinzuzufügende Provider.

        Returns:
            Self für Method-Chaining.
        """
        self._providers.append(provider)
        return self

    def get(self, key: str, default: T | None = None) -> T | None:
        """Sucht in allen Providern nach dem Secret."""
        for provider in self._providers:
            value = provider.get(key)
            if value is not None:
                return value
        return default

    def get_secret(self, key: str) -> SecretValue | None:
        """Sucht in allen Providern nach dem Secret mit Metadaten."""
        for provider in self._providers:
            secret = provider.get_secret(key)
            if secret is not None:
                return secret
        return None

    def exists(self, key: str) -> bool:
        """Prüft in allen Providern."""
        return any(p.exists(key) for p in self._providers)

    def list_keys(self) -> list[str]:
        """Sammelt Schlüssel aus allen Providern."""
        keys: set[str] = set()
        for provider in self._providers:
            try:
                keys.update(provider.list_keys())
            except NotImplementedError:
                pass
        return sorted(keys)


class CachingSecretsProvider(SecretsProvider):
    """
    Wrapper der Secrets cached.

    Reduziert Zugriffe auf den zugrunde liegenden Provider
    durch zeitbasiertes Caching.
    """

    def __init__(
        self,
        provider: SecretsProvider,
        ttl_seconds: float = 300.0,
        name: str = "",
    ) -> None:
        """
        Initialisiert den Caching-Wrapper.

        Args:
            provider: Der zugrunde liegende Provider.
            ttl_seconds: Cache-TTL in Sekunden.
            name: Name des Providers.
        """
        super().__init__(name or f"Cached({provider.name})")
        self._provider = provider
        self._ttl_seconds = ttl_seconds
        self._cache_times: dict[str, datetime] = {}

    def _is_cache_valid(self, key: str) -> bool:
        """Prüft, ob der Cache-Eintrag noch gültig ist."""
        if key not in self._cache or key not in self._cache_times:
            return False

        from datetime import timedelta
        age = datetime.now() - self._cache_times[key]
        return age.total_seconds() < self._ttl_seconds

    def get(self, key: str, default: T | None = None) -> T | None:
        """Holt aus Cache oder Provider."""
        if self._cache_enabled and self._is_cache_valid(key):
            return self._cache[key].value

        value = self._provider.get(key, default)

        if value is not None and self._cache_enabled:
            self._cache[key] = SecretValue(
                key=key,
                value=value,
                source=self._provider.name,
            )
            self._cache_times[key] = datetime.now()

        return value

    def get_secret(self, key: str) -> SecretValue | None:
        """Holt Secret mit Metadaten aus Cache oder Provider."""
        if self._cache_enabled and self._is_cache_valid(key):
            return self._cache[key]

        secret = self._provider.get_secret(key)

        if secret is not None and self._cache_enabled:
            self._cache[key] = secret
            self._cache_times[key] = datetime.now()

        return secret

    def clear_cache(self) -> None:
        """Löscht den Cache."""
        self._cache.clear()
        self._cache_times.clear()

    def exists(self, key: str) -> bool:
        """Prüft Existenz."""
        if self._cache_enabled and key in self._cache:
            return True
        return self._provider.exists(key)

    def list_keys(self) -> list[str]:
        """Listet alle Schlüssel."""
        return self._provider.list_keys()