codeBOKER/customer_service

1from collections import defaultdict, deque2from time import monotonic3 4from fastapi import HTTPException, Request5 6from config import TELEGRAM_WEBHOOK_SECRET7 8FAILED_ATTEMPT_WINDOW_SECONDS = 609FAILED_ATTEMPT_LIMIT = 510BLOCK_DURATION_SECONDS = 15 * 6011 12_failed_secret_attempts: dict[str, deque[float]] = defaultdict(deque)13_blocked_clients: dict[str, float] = {}14 15 16def _get_client_key(request: Request) -> str:17    forwarded_for = request.headers.get("x-forwarded-for")18    if forwarded_for:19        return forwarded_for.split(",")[0].strip()20 21    if request.client and request.client.host:22        return request.client.host23 24    return "unknown"25 26 27def _prune_failed_attempts(client_key: str, now: float) -> deque[float]:28    attempts = _failed_secret_attempts[client_key]29    cutoff = now - FAILED_ATTEMPT_WINDOW_SECONDS30    while attempts and attempts[0] < cutoff:31        attempts.popleft()32    return attempts33 34 35def validate_webhook_secret(request: Request, secret_header: str | None) -> None:36    if not TELEGRAM_WEBHOOK_SECRET:37        raise HTTPException(status_code=500, detail="Webhook secret is not configured")38 39    client_key = _get_client_key(request)40    now = monotonic()41    blocked_until = _blocked_clients.get(client_key)42    if blocked_until and now < blocked_until:43        raise HTTPException(status_code=429, detail="Too many requests")44 45    if secret_header != TELEGRAM_WEBHOOK_SECRET:46        attempts = _prune_failed_attempts(client_key, now)47        attempts.append(now)48        if len(attempts) >= FAILED_ATTEMPT_LIMIT:49            _blocked_clients[client_key] = now + BLOCK_DURATION_SECONDS50            attempts.clear()51        raise HTTPException(status_code=403, detail="Forbidden")52 53    _blocked_clients.pop(client_key, None)54    _failed_secret_attempts.pop(client_key, None)55