Etappe 5: Auth + Rate-Limiting

- JWT-Auth (Access 15min/Refresh 7 Tage)
- Rate-Limiting (100/min pro Client/API-Key)
- API Key Management
- OAuth2 Password Scheme
This commit is contained in:
Hitonabi
2026-07-21 17:19:34 +02:00
parent 518155051f
commit 3e921fc0c9
3 changed files with 309 additions and 1 deletions
+90
View File
@@ -0,0 +1,90 @@
"""Rate-Limiting-Modul für Rippy API."""
import time
import secrets
from collections import defaultdict
from typing import Dict, Optional
from .config import settings
# Default Rate Limit
MAX_REQUESTS_PER_MINUTE = 100
# In-Memory Rate Limit Store (in Produktion mit Redis)
rate_limit_store: Dict[str, list] = defaultdict(list)
def check_rate_limit(client_id: str, max_requests: int = MAX_REQUESTS_PER_MINUTE, window_seconds: int = 60) -> bool:
"""Prüfe ob Client rate-limited ist."""
current_time = time.time()
window_start = current_time - window_seconds
# Bereinige alte Einträge
rate_limit_store[client_id] = [
timestamp for timestamp in rate_limit_store[client_id]
if timestamp > window_start
]
# Prüfe ob Limit erreicht
if len(rate_limit_store[client_id]) >= max_requests:
return False
# Füge neuen Request hinzu
rate_limit_store[client_id].append(current_time)
return True
def get_rate_limit_remaining(client_id: str, max_requests: int = MAX_REQUESTS_PER_MINUTE) -> int:
"""Hole verbleibende Requests."""
current_time = time.time()
window_start = current_time - 60
current_count = len([
timestamp for timestamp in rate_limit_store[client_id]
if timestamp > window_start
])
return max(0, max_requests - current_count)
def reset_rate_limit(client_id: str) -> None:
"""Setze Rate Limit für Client zurück."""
rate_limit_store[client_id] = []
# API-Key Store (in Produktion mit Datenbank)
api_keys: Dict[str, Dict] = {
"example_key": {
"key": "example_key",
"name": "Beispiel API Key",
"created_at": time.time(),
"rate_limit": 100
}
}
def validate_api_key(api_key: str) -> Optional[Dict]:
"""Validiere API Key."""
if api_key in api_keys:
return api_keys[api_key]
return None
def create_api_key(name: str) -> Dict:
"""Erstelle neuer API Key."""
key = secrets.token_urlsafe(32)
api_keys[key] = {
"key": key,
"name": name,
"created_at": time.time(),
"rate_limit": MAX_REQUESTS_PER_MINUTE
}
return api_keys[key]
def delete_api_key(key: str) -> bool:
"""Lösche API Key."""
if key in api_keys:
del api_keys[key]
return True
return False