pelmen/baton
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

kin: BATON-SEC-002-backend_dev

Browse source
This commit is contained in:
Gros Frumos 2026-03-21 07:36:33 +02:00
parent 3483b71fcb
commit 63e99d87ef
2 changed files with 28 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

27
backend/middleware.py
View file

@ -14,6 +14,17 @@ _bearer = HTTPBearer(auto_error=False)
_RATE_LIMIT = 5
_RATE_WINDOW = 600 # 10 minutes
_SIGNAL_RATE_LIMIT = 10
_SIGNAL_RATE_WINDOW = 60 # 1 minute
def _get_client_ip(request: Request) -> str:
return (
request.headers.get("X-Real-IP")
or request.headers.get("X-Forwarded-For", "").split(",")[0].strip()
or (request.client.host if request.client else "unknown")
)
async def verify_webhook_secret(
x_telegram_bot_api_secret_token: str = Header(default=""),
@ -35,7 +46,7 @@ async def verify_admin_token(
async def rate_limit_register(request: Request) -> None:
counters = request.app.state.rate_counters
client_ip = request.client.host if request.client else "unknown"
client_ip = _get_client_ip(request)
now = time.time()
count, window_start = counters.get(client_ip, (0, now))
if now - window_start >= _RATE_WINDOW:
@ -45,3 +56,17 @@ async def rate_limit_register(request: Request) -> None:
counters[client_ip] = (count, window_start)
if count > _RATE_LIMIT:
raise HTTPException(status_code=429, detail="Too Many Requests")
async def rate_limit_signal(request: Request) -> None:
counters = request.app.state.rate_counters
key = f"sig:{_get_client_ip(request)}"
now = time.time()
count, window_start = counters.get(key, (0, now))
if now - window_start >= _SIGNAL_RATE_WINDOW:
count = 0
window_start = now
count += 1
counters[key] = (count, window_start)
if count > _SIGNAL_RATE_LIMIT:
raise HTTPException(status_code=429, detail="Too Many Requests")