34 lines
981 B
Python
34 lines
981 B
Python
from __future__ import annotations
|
|
|
|
import secrets
|
|
import time
|
|
|
|
from fastapi import Header, HTTPException, Request
|
|
|
|
from backend import config
|
|
|
|
_RATE_LIMIT = 5
|
|
_RATE_WINDOW = 600 # 10 minutes
|
|
|
|
|
|
async def verify_webhook_secret(
|
|
x_telegram_bot_api_secret_token: str = Header(default=""),
|
|
) -> None:
|
|
if not secrets.compare_digest(
|
|
x_telegram_bot_api_secret_token, config.WEBHOOK_SECRET
|
|
):
|
|
raise HTTPException(status_code=403, detail="Forbidden")
|
|
|
|
|
|
async def rate_limit_register(request: Request) -> None:
|
|
counters = request.app.state.rate_counters
|
|
client_ip = request.client.host if request.client else "unknown"
|
|
now = time.time()
|
|
count, window_start = counters.get(client_ip, (0, now))
|
|
if now - window_start >= _RATE_WINDOW:
|
|
count = 0
|
|
window_start = now
|
|
count += 1
|
|
counters[client_ip] = (count, window_start)
|
|
if count > _RATE_LIMIT:
|
|
raise HTTPException(status_code=429, detail="Too Many Requests")
|