Hitonabi
3e921fc0c9
- JWT-Auth (Access 15min/Refresh 7 Tage) - Rate-Limiting (100/min pro Client/API-Key) - API Key Management - OAuth2 Password Scheme
91 lines
2.3 KiB
Python
91 lines
2.3 KiB
Python
"""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
|