"""Rate-Limiting-Modul für Rippy API (pro Client-IP). Der API-Key-Store, der hier früher lebte, ist mit dem Auth-Rückbau (Commander-Entscheid 24.07.2026) entfernt — Heimnetz-only, siehe KONZEPT §10. """ import time from collections import defaultdict from typing import Dict # 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] = []