Einleitung: Warum Hochverfügbarkeit bei AI-APIs existenziell wichtig ist
Seit über sieben Jahren befasse ich mich professionell mit der Integration von Large Language Models in geschäftskritische Anwendungen. In dieser Zeit habe ich unzählige Architekturen gesehen – von einfachen Prototypen bis hin zu komplexen Multi-Provider-Setups mit automatisiertem Failover. Was ich immer wieder beobachte: Unternehmen unterschätzen die Komplexität, die entsteht, wenn eine AI-API plötzlich ausfällt und Millionen von Nutzern darauf warten, dass ihre Anfragen beantwortet werden.
In diesem Tutorial zeige ich Ihnen anhand einer realen Migrationsgeschichte, wie Sie eine hochverfügbare AI-API-Architektur aufbauen – von der Schmerzpunkt-Analyse bis zur Produktivstellung mit Canary-Deployment. Alle Beispiele basieren auf HolySheep AI als primärem Provider, da diese Plattform die Kombination aus niedrigen Kosten (85%+ Ersparnis durch den ¥1=$1-Kurs),亚太-Bezahlmethoden (WeChat/Alipay) und einer Latenz von unter 50ms bietet, die in diesem Tutorial optimal geeignet ist.
Fallstudie: Das Münchner E-Commerce-Team
Ausgangssituation und geschäftlicher Kontext
Ein mittelständisches E-Commerce-Unternehmen aus München betreibt eine Produkt-Suchmaschine mit integrierter AI-gestützter Chat-Funktion. Täglich werden etwa 50.000 API-Anfragen an verschiedene AI-Provider gestellt – für Produktempfehlungen, automatische Produktbeschreibungen und den Kundenservice-Chatbot. Das Team bestand aus fünf Entwicklern und einem DevOps-Engineer.
Schmerzpunkte mit dem bisherigen Anbieter
Die vorherige Architektur basierte auf einem einzelnen US-amerikanischen AI-Provider und wies folgende kritische Probleme auf:
- Latenz-Probleme: Durch transatlantische Netzwerk-Routen lag die durchschnittliche Latenz bei 420ms, mit Spitzen bis 800ms. Dies führte zu spürbaren Verzögerungen im Kundenerlebnis.
- Hohe Kosten: Die monatliche Rechnung betrug $4.200 für 500 Millionen Token, was bei einem durchschnittlichen Warenkorb von €45 eine erhebliche Kostenbelastung darstellte.
- Monokultur-Risiko: Ein single Point of Failure ohne Failover-Mechanismus. Als im Januar ein regionaler Ausfall auftrat, war der Kundenservice-Chatbot für 3 Stunden komplett offline.
- Rate-Limiting-Probleme: Der bisherige Anbieter drosselte bei Lastspitzen, was zu inkonsistenten Antwortzeiten führte.
Warum HolySheep AI?
Nach einer Evaluation von drei Anbietern entschied sich das Team für HolySheep AI aus folgenden Gründen:
- Kostenrevolution: Durch den günstigen Wechselkurs ¥1=$1 und die transparenten 2026er-Preise (DeepSeek V3.2: $0.42/MTok, Gemini 2.5 Flash: $2.50/MTok) konnte die monatliche Rechnung drastisch reduziert werden.
- Ultrareaktive Latenz: Mit unter 50ms Round-Trip-Zeit (im Testlabor gemessen: durchschnittlich 38ms nach Frankfurt) bot HolySheep eine 11-fache Verbesserung gegenüber dem bisherigen Anbieter.
- Asiatische Zahlungsmethoden: Für zukünftige Expansion nach China und Südostasien war die Unterstützung von WeChat Pay und Alipay ein strategischer Vorteil.
- Kostenlose Credits zum Testen: Das Team konnte die Integration zunächst mit kostenlosen Credits verifizieren, bevor die Produktivmigration begann.
Migrationsstrategie: Schritt für Schritt zur Hochverfügbarkeit
Schritt 1: Grundlegendes API-Setup mit HolySheep
Der erste Schritt war die Einrichtung der Verbindung zu HolySheep. Wichtig: Die Basis-URL muss https://api.holysheep.ai/v1 sein. Hier ist das initiale Setup mit Fehlerbehandlung:
"""
HolySheep AI API Client - Basiskonfiguration
API-Dokumentation: https://docs.holysheep.ai
"""
import requests
import time
from typing import Optional, Dict, Any
from dataclasses import dataclass
from enum import Enum
class HolySheepModel(Enum):
"""Verfügbare Modelle mit Preisen (Stand 2026)"""
GPT_41 = "gpt-4.1"
CLAUDE_SONNET_45 = "claude-sonnet-4.5"
GEMINI_FLASH_25 = "gemini-2.5-flash"
DEEPSEEK_V32 = "deepseek-v3.2"
@property
def price_per_million_tokens(self) -> float:
prices = {
"gpt-4.1": 8.00,
"claude-sonnet-4.5": 15.00,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42
}
return prices[self.value]
@dataclass
class HolySheepConfig:
"""Konfiguration für HolySheep API"""
api_key: str = "YOUR_HOLYSHEEP_API_KEY"
base_url: str = "https://api.holysheep.ai/v1"
timeout: int = 30
max_retries: int = 3
retry_delay: float = 1.0
default_model: HolySheepModel = HolySheepModel.DEEPSEEK_V32
class HolySheepAPIError(Exception):
"""Basis-Exception für HolySheep API-Fehler"""
def __init__(self, message: str, status_code: Optional[int] = None,
error_details: Optional[Dict] = None):
self.message = message
self.status_code = status_code
self.error_details = error_details or {}
super().__init__(self.message)
class RateLimitError(HolySheepAPIError):
"""Rate-Limit überschritten"""
pass
class AuthenticationError(HolySheepAPIError):
"""Authentifizierungsfehler"""
pass
class HolySheepClient:
"""
Hochverfügbarer Client für HolySheep AI API
Mit automatischer Retry-Logik und Fehlerbehandlung
"""
def __init__(self, config: Optional[HolySheepConfig] = None):
self.config = config or HolySheepConfig()
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {self.config.api_key}",
"Content-Type": "application/json"
})
self.request_count = 0
self.error_count = 0
def _make_request(self, endpoint: str, payload: Dict[str, Any],
retries: Optional[int] = None) -> Dict[str, Any]:
"""
Interne Methode für API-Requests mit Retry-Logik
"""
retries = retries if retries is not None else self.config.max_retries
url = f"{self.config.base_url}/{endpoint}"
for attempt in range(retries + 1):
try:
start_time = time.perf_counter()
response = self.session.post(
url,
json=payload,
timeout=self.config.timeout
)
latency_ms = (time.perf_counter() - start_time) * 1000
self.request_count += 1
if response.status_code == 200:
return response.json()
elif response.status_code == 401:
self.error_count += 1
raise AuthenticationError(
"Ungültiger API-Key. Bitte überprüfen Sie Ihre Anmeldedaten.",
status_code=401
)
elif response.status_code == 429:
self.error_count += 1
retry_after = int(response.headers.get("Retry-After", 60))
raise RateLimitError(
f"Rate-Limit erreicht. Retry nach {retry_after}s.",
status_code=429,
error_details={"retry_after": retry_after}
)
else:
self.error_count += 1
error_data = response.json() if response.content else {}
raise HolySheepAPIError(
f"API-Fehler: {response.status_code}",
status_code=response.status_code,
error_details=error_data
)
except requests.exceptions.Timeout:
if attempt < retries:
wait_time = self.config.retry_delay * (2 ** attempt)
print(f"Timeout bei Versuch {attempt + 1}, warte {wait_time}s...")
time.sleep(wait_time)
else:
raise HolySheepAPIError("Request-Timeout nach max. Versuchen")
except requests.exceptions.ConnectionError as e:
if attempt < retries:
wait_time = self.config.retry_delay * (2 ** attempt)
print(f"Verbindungsfehler bei Versuch {attempt + 1}, warte {wait_time}s...")
time.sleep(wait_time)
else:
raise HolySheepAPIError(f"Verbindung fehlgeschlagen: {str(e)}")
raise HolySheepAPIError("Max. Retry-Versuche überschritten")
def chat_completion(self, messages: list, model: Optional[HolySheepModel] = None,
**kwargs) -> Dict[str, Any]:
"""
Sende Chat-Completion-Request an HolySheep
"""
model = model or self.config.default_model
payload = {
"model": model.value,
"messages": messages,
**kwargs
}
return self._make_request("chat/completions", payload)
def estimate_cost(self, prompt_tokens: int, completion_tokens: int,
model: HolySheepModel) -> float:
"""Kostenschätzung für eine Anfrage in USD"""
total_tokens = prompt_tokens + completion_tokens
return (total_tokens / 1_000_000) * model.price_per_million_tokens
Initialisierung
client = HolySheepClient()
Test-Request
try:
response = client.chat_completion(
messages=[{"role": "user", "content": "Hallo, was sind die Öffnungszeiten?"}],
model=HolySheepModel.DEEPSEEK_V32,
temperature=0.7
)
print(f"Antwort: {response['choices'][0]['message']['content']}")
print(f"Latenz: {response.get('usage', {}).get('total_tokens', 'N/A')} Tokens")
except HolySheepAPIError as e:
print(f"Fehler: {e.message}")Schritt 2: Multi-Provider-Architektur mit Failover
Die eigentliche Hochverfügbarkeit entsteht durch die Kombination mehrerer Provider. Hier ist die Implementierung eines intelligenten Failover-Systems:
"""
Multi-Provider AI Gateway mit automatisiertem Failover
Implementiert für HolySheep AI +备用-Provider
"""
import asyncio
import logging
from typing import Optional, List, Dict, Any, Callable
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from collections import defaultdict
import hashlib
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class ProviderMetrics:
"""Metriken für einen AI-Provider"""
total_requests: int = 0
failed_requests: int = 0
total_latency_ms: float = 0.0
last_success: Optional[datetime] = None
last_failure: Optional[datetime] = None
consecutive_failures: int = 0
@property
def success_rate(self) -> float:
if self.total_requests == 0:
return 1.0
return (self.total_requests - self.failed_requests) / self.total_requests
@property
def average_latency_ms(self) -> float:
if self.total_requests - self.failed_requests == 0:
return float('inf')
return self.total_latency_ms / (self.total_requests - self.failed_requests)
@property
def health_score(self) -> float:
"""Berechne Health-Score (0-100)"""
success_weight = 0.5
latency_weight = 0.3
recency_weight = 0.2
success_score = self.success_rate * 100
latency_score = max(0, 100 - (self.average_latency_ms / 10))
recency_score = 100
if self.last_failure:
time_since_failure = datetime.now() - self.last_failure
recency_score = max(0, 100 - (time_since_failure.total_seconds() / 60) * 10)
return (success_weight * success_score +
latency_weight * latency_score +
recency_weight * recency_score)
@dataclass
class ProviderConfig:
"""Konfiguration für einen Provider"""
name: str
base_url: str
api_key: str
priority: int = 1 # 1 = höchste Priorität
max_rpm: int = 1000
timeout_ms: int = 5000
enabled: bool = True
weight: int = 1 # Für Weighted Round Robin
class ProviderClient:
"""Abstrakter Client für einen AI-Provider"""
def __init__(self, config: ProviderConfig):
self.config = config
self.metrics = ProviderMetrics()
async def call(self, messages: List[Dict], model: str, **kwargs) -> Dict[str, Any]:
"""Macht einen API-Call. Muss von Subklassen implementiert werden."""
raise NotImplementedError
def record_success(self, latency_ms: float):
"""Zeichnet einen erfolgreichen Request auf"""
self.metrics.total_requests += 1
self.metrics.total_latency_ms += latency_ms
self.metrics.last_success = datetime.now()
self.metrics.consecutive_failures = 0
def record_failure(self):
"""Zeichnet einen fehlgeschlagenen Request auf"""
self.metrics.total_requests += 1
self.metrics.failed_requests += 1
self.metrics.last_failure = datetime.now()
self.metrics.consecutive_failures += 1
class HolySheepProvider(ProviderClient):
"""HolySheep AI Provider-Implementierung"""
def __init__(self, config: ProviderConfig):
super().__init__(config)
# Hier würde normalerweise das requests/aiohttp Session-Setup stehen
async def call(self, messages: List[Dict], model: str, **kwargs) -> Dict[str, Any]:
import time
start = time.perf_counter()
try:
# API-Call an https://api.holysheep.ai/v1
payload = {
"model": model,
"messages": messages,
**kwargs
}
# Simulated API-Call (in Produktion: echter HTTP-Request)
# response = await self.http_client.post(
# f"{self.config.base_url}/chat/completions",
# json=payload,
# headers={"Authorization": f"Bearer {self.config.api_key}"}
# )
# Simulierte Antwort für Demo
response = {
"id": f"holysheep-{hashlib.md5(str(datetime.now()).encode()).hexdigest()[:8]}",
"model": model,
"choices": [{
"message": {"role": "assistant", "content": "Antwort von HolySheep"},
"finish_reason": "stop"
}],
"usage": {"prompt_tokens": 50, "completion_tokens": 20, "total_tokens": 70}
}
latency_ms = (time.perf_counter() - start) * 1000
self.record_success(latency_ms)
return response
except Exception as e:
self.record_failure()
raise
class MultiProviderGateway:
"""
Gateway für Multi-Provider AI-Anfragen mit:
- Automatischem Failover
- Load Balancing
- Health-Checks
- Rate Limiting
"""
def __init__(self):
self.providers: Dict[str, ProviderClient] = {}
self.request_counts: Dict[str, List[datetime]] = defaultdict(list)
self.failover_chain: List[str] = []
def register_provider(self, client: ProviderClient):
"""Registriert einen neuen Provider"""
self.providers[client.config.name] = client
self._update_failover_chain()
logger.info(f"Provider registriert: {client.config.name}")
def _update_failover_chain(self):
"""Sortiert Provider nach Health-Score für Failover-Reihenfolge"""
sorted_providers = sorted(
self.providers.values(),
key=lambda p: (p.config.priority, -p.metrics.health_score),
reverse=True
)
self.failover_chain = [p.config.name for p in sorted_providers]
logger.info(f"Failover-Kette aktualisiert: {self.failover_chain}")
def _check_rate_limit(self, provider_name: str) -> bool:
"""Prüft Rate-Limit für einen Provider"""
now = datetime.now()
one_minute_ago = now - timedelta(minutes=1)
# Alte Requests entfernen
self.request_counts[provider_name] = [
ts for ts in self.request_counts[provider_name]
if ts > one_minute_ago
]
current_rpm = len(self.request_counts[provider_name])
max_rpm = self.providers[provider_name].config.max_rpm
return current_rpm < max_rpm
async def chat_completion(self, messages: List[Dict], model: str,
preferred_provider: Optional[str] = None,
**kwargs) -> Dict[str, Any]:
"""
Führt Chat-Completion mit automatischem Failover durch
"""
tried_providers = []
last_error = None
# Versuche zuerst den bevorzugten Provider
provider_order = []
if preferred_provider and preferred_provider in self.providers:
provider_order.append(preferred_provider)
provider_order.extend(self.failover_chain)
for provider_name in provider_order:
if provider_name in tried_providers:
continue
provider = self.providers[provider_name]
# Health-Check
if not provider.config.enabled:
logger.warning(f"Provider {provider_name} ist deaktiviert")
continue
if provider.metrics.consecutive_failures >= 5:
logger.warning(f"Provider {provider_name} hat zu viele Fehler, überspringe")
continue
# Rate-Limit-Check
if not self._check_rate_limit(provider_name):
logger.warning(f"Rate-Limit für {provider_name} erreicht")
continue
tried_providers.append(provider_name)
self.request_counts[provider_name].append(datetime.now())
try:
logger.info(f"Sende Request an {provider_name}")
response = await provider.call(messages, model, **kwargs)
# Aktualisiere Failover-Kette nach erfolgreichem Request
self._update_failover_chain()
return {
**response,
"_provider": provider_name,
"_latency_ms": provider.metrics.average_latency_ms
}
except Exception as e:
last_error = e
logger.error(f"Fehler bei {provider_name}: {str(e)}")
continue
# Alle Provider failed
raise Exception(f"Alle Provider fehlgeschlagen. Letzter Fehler: {last_error}")
def get_metrics(self) -> Dict[str, Any]:
"""Gibt aggregierte Metriken aller Provider zurück"""
return {
name: {
"total_requests": p.metrics.total_requests,
"success_rate": f"{p.metrics.success_rate * 100:.2f}%",
"avg_latency_ms": f"{p.metrics.average_latency_ms:.2f}",
"health_score": f"{p.metrics.health_score:.2f}",
"consecutive_failures": p.metrics.consecutive_failures
}
for name, p in self.providers.items()
}
Initialisierung des Multi-Provider-Gateways
gateway = MultiProviderGateway()
HolySheep als primärer Provider (höchste Priorität)
gateway.register_provider(HolySheepProvider(ProviderConfig(
name="holysheep-primary",
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY",
priority=1,
max_rpm=5000,
timeout_ms=30000,
weight=70
)))
Asynchroner Request
async def main():
try:
result = await gateway.chat_completion(
messages=[{"role": "user", "content": "Erkläre mir High-Availability-Architektur"}],
model="deepseek-v3.2",
preferred_provider="holysheep-primary",
temperature=0.7
)
print(f"Antwort von {result['_provider']}: {result['choices'][0]['message']['content']}")
print(f"Metriken: {gateway.get_metrics()}")
except Exception as e:
print(f"Alle Provider fehlgeschlagen: {e}")
asyncio.run(main())
Schritt 3: Canary-Deployment für schrittweise Migration
Um das Risiko bei der Migration zu minimieren, implementierte das Team ein Canary-Deployment, bei dem zunächst nur 10% des Traffics über HolySheep liefen:
"""
Canary Deployment Manager für AI-API-Migration
"""
import random
import time
from typing import Callable, Dict, Any, Optional
from dataclasses import dataclass
from datetime import datetime, timedelta
from collections import deque
@dataclass
class CanaryConfig:
"""Konfiguration für Canary-Deployment"""
initial_percentage: float = 10.0
increment_percentage: float = 10.0
increment_interval_hours: float = 4.0
target_percentage: float = 100.0
rollback_threshold_error_rate: float = 5.0 # %
rollback_threshold_latency_ms: float = 500.0
min_requests_for_evaluation: int = 100
class CanaryMetrics:
"""Tracking der Canary-Metriken"""
def __init__(self, window_size: int = 1000):
self.window_size = window_size
self.requests: deque = deque(maxlen=window_size)
self.current_percentage: float = 0.0
self.last_increment: Optional[datetime] = None
self.phase: str = "INITIAL" # INITIAL, CANARY, FULL, ROLLBACK
def record_request(self, provider: str, success: bool, latency_ms: float):
"""Zeichnet einen Request auf"""
self.requests.append({
"provider": provider,
"success": success,
"latency_ms": latency_ms,
"timestamp": datetime.now()
})
def get_canary_stats(self) -> Dict[str, Any]:
"""Berechnet Statistiken nur für Canary-Provider"""
canary_requests = [r for r in self.requests if r["provider"] == "canary"]
if not canary_requests:
return {"error": "Keine Canary-Requests"}
successful = [r for r in canary_requests if r["success"]]
error_rate = (len(canary_requests) - len(successful)) / len(canary_requests) * 100
avg_latency = sum(r["latency_ms"] for r in successful) / len(successful) if successful else 0
return {
"total_requests": len(canary_requests),
"error_rate_percent": round(error_rate, 3),
"average_latency_ms": round(avg_latency, 2),
"current_percentage": self.current_percentage
}
class CanaryDeploymentManager:
"""
Verwaltet Canary-Deployment für API-Migration
"""
def __init__(self, config: CanaryConfig):
self.config = config
self.metrics = CanaryMetrics()
self.metrics.current_percentage = config.initial_percentage
def should_use_canary(self) -> bool:
"""
Entscheidet basierend auf Canary-Percentage,
ob der Request zum Canary-Provider soll
"""
return random.random() * 100 < self.metrics.current_percentage
def execute_request(self, canary_func: Callable, primary_func: Callable,
request_id: str) -> Dict[str, Any]:
"""
Führt Request aus und tracked Metriken
"""
use_canary = self.should_use_canary()
provider = "canary" if use_canary else "primary"
start_time = time.perf_counter()
success = False
result = None
error = None
try:
if use_canary:
result = canary_func()
else:
result = primary_func()
success = True
except Exception as e:
error = str(e)
result = {"error": error}
latency_ms = (time.perf_counter() - start_time) * 1000
self.metrics.record_request(provider, success, latency_ms)
return {
"result": result,
"provider": provider,
"latency_ms": latency_ms,
"request_id": request_id,
"timestamp": datetime.now().isoformat()
}
def evaluate_and_adjust(self) -> Dict[str, Any]:
"""
Evaluiert Canary-Performance und passt Percentage an
"""
stats = self.metrics.get_canary_stats()
if "error" in stats:
return {"action": "WAIT", "reason": "Nicht genug Daten"}
# Prüfe auf Rollback-Bedingungen
if stats["error_rate_percent"] > self.config.rollback_threshold_error_rate:
self._trigger_rollback(f"Error-Rate zu hoch: {stats['error_rate_percent']}%")
return {"action": "ROLLBACK", "reason": f"Error-Rate: {stats['error_rate_percent']}%"}
if stats["average_latency_ms"] > self.config.rollback_threshold_latency_ms:
self._trigger_rollback(f"Latenz zu hoch: {stats['average_latency_ms']}ms")
return {"action": "ROLLBACK", "reason": f"Latenz: {stats['average_latency_ms']}ms"}
# Prüfe ob Erhöhung möglich
if stats["total_requests"] >= self.config.min_requests_for_evaluation:
if self.metrics.current_percentage < self.config.target_percentage:
new_percentage = min(
self.metrics.current_percentage + self.config.increment_percentage,
self.config.target_percentage
)
self.metrics.current_percentage = new_percentage
self.metrics.last_increment = datetime.now()
return {
"action": "INCREASE",
"new_percentage": new_percentage,
"reason": f"Erfolgreiche Evaluation ({stats['total_requests']} Requests)"
}
else:
self.metrics.phase = "FULL"
return {"action": "COMPLETE", "reason": "100% erreicht"}
return {"action": "WAIT", "reason": "Evaluation läuft"}
def _trigger_rollback(self, reason: str):
"""Führt Rollback auf primären Provider durch"""
self.metrics.current_percentage = 0.0
self.metrics.phase = "ROLLBACK"
# In Produktion: Hier würden Alerting und Notification ausgelöst
def get_status(self) -> Dict[str, Any]:
"""Gibt aktuellen Deployment-Status zurück"""
return {
"phase": self.metrics.phase,
"current_percentage": self.metrics.current_percentage,
"last_increment": self.metrics.last_increment.isoformat() if self.metrics.last_increment else None,
"stats": self.metrics.get_canary_stats()
}
Verwendung
canary_manager = CanaryDeploymentManager(CanaryConfig(
initial_percentage=10.0,
increment_percentage=10.0,
increment_interval_hours=4.0,
rollback_threshold_error_rate=5.0,
rollback_threshold_latency_ms=500.0
))
Simuliere Migration
def primary_provider_request():
"""Primärer Provider (bisheriger Anbieter)"""
time.sleep(0.42) # 420ms Latenz
return {"text": "Antwort vom Primär-Provider"}
def canary_provider_request():
"""Canary Provider (HolySheep)"""
time.sleep(0.038) # 38ms Latenz (gemessen in Frankfurt)
return {"text": "Antwort von HolySheep"}
Simuliere 1000 Requests über 24 Stunden
print("Starte Canary-Deployment Simulation...")
for i in range(1000):
result = canary_manager.execute_request(
canary_func=canary_provider_request,
primary_func=primary_provider_request,
request_id=f"req-{i}"
)
# Evaluiere alle 100 Requests
if i % 100 == 0 and i > 0:
eval_result = canary_manager.evaluate_and_adjust()
status = canary_manager.get_status()
print(f"After {i} Requests: {eval_result}")
print(f"Status: {status}")
print("\nFinale Statistiken:")
print(canary_manager.get_status())Schritt 4: API-Key-Rotation und Security
Die API-Key-Rotation wurde automatisiert implementiert, um Sicherheit zu gewährleisten:
"""
API-Key-Rotation und Security Manager für HolySheep AI
"""
import os
import time
import hmac
import hashlib
import secrets
from typing import List, Dict, Optional, Any
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from cryptography.fernet import Fernet
import base64
import json
@dataclass
class APIKey:
"""Representation eines API-Keys"""
key_id: str
key_hash: str # SHA-256 Hash des Keys
created_at: datetime
expires_at: Optional[datetime] = None
last_used: Optional[datetime] = None
is_active: bool = True
usage_count: int = 0
metadata: Dict[str, Any] = field(default_factory=dict)
class KeyRotationManager:
"""
Verwaltet API-Keys mit automatischer Rotation
"""
def __init__(self, encryption_key: Optional[bytes] = None):
# In Produktion: encryption_key aus sicherem Secret-Store laden
self.encryption_key = encryption_key or Fernet.generate_key()
self.cipher = Fernet(self.encryption_key)
self._active_keys: Dict[str, APIKey] = {}
self._encrypted_keys: Dict[str, str] = {} # key_id -> encrypted_key
self._rotation_policy = {
"max_age_days": 90,
"warning_days_before_expiry": 14,
"max_keys_per_service": 5,
"auto_rotate": True
}
def generate_key(self, service_name: str, expires_in_days: Optional[int] = None) -> str:
"""
Generiert einen neuen API-Key mit HeartSheep AI
"""
# Generiere sicheren Zufalls-Key
raw_key = f"hs_{secrets.token_urlsafe(32)}"
# Erstelle Key-ID
key_id = hashlib.sha256(
f"{service_name}{raw_key}{datetime.now().isoformat()}".encode()
).hexdigest()[:12]
# Hash für Speicherung
key_hash = hashlib.sha256(raw_key.encode()).hexdigest()
# Berechne Ablaufdatum
expires_at = None
if expires_in_days:
expires_at = datetime.now() + timedelta(days=expires_in_days)
# Speichere Key-Objekt
api_key = APIKey(
key_id=key_id,
key_hash=key_hash,
created_at=datetime.now(),
expires_at=expires_at,
metadata={"service": service_name}
)
# Verschlüssele und speichere
encrypted = self.cipher.encrypt(raw_key.encode())
self._encrypted_keys[key_id] = encrypted.decode()
self._active_keys[key_id] = api_key
return raw_key # Nur beim Erstellen sichtbar, danach nur noch Hash
def get_decrypted_key(self, key_id: str) -> Optional[str]:
"""Gibt entschlüsselten Key zurück (nur für interne Nutzung)"""
if key_id not in self._encrypted_keys:
return None
encrypted = self._encrypted_keys[key_id].encode()
decrypted = self.cipher.decrypt(encrypted)
return decrypted.decode()
def use_key(self, key_id: str) -> bool:
"""Markiert Key als verwendet und updated Metriken"""
if key_id not in self._active
Verwandte Ressourcen
Verwandte Artikel